diff --git a/docs/ATTACHMENT-RECOVERY.md b/docs/ATTACHMENT-RECOVERY.md new file mode 100644 index 0000000..be5459c --- /dev/null +++ b/docs/ATTACHMENT-RECOVERY.md @@ -0,0 +1,37 @@ +# Attachment cleanup recovery + +Deletion writes a standard `collaboration_attachment_cleanup` row and removes attachment metadata in the same backend transaction. Only after commit does it delete the stored object, then acknowledge the cleanup row. PostgreSQL rejection of metadata deletion leaves both the original row and bytes intact. A provider failure or lost callback leaves the cleanup locator available after restart. The service returns a failure when immediate cleanup fails; the logical attachment deletion may already have committed. + +The cleanup table is visible in the Collaboration app to administrators. Direct table mutations are denied, including to administrators, so callers cannot forge deletion targets. Recovery runs through `collaborationRetryAttachmentCleanup`, guarded for system sessions or `collaboration.admin`. With the host's configured scheduler, it retries a bounded batch of 100 every minute. Kof22 already supplies a scheduler name. Without a scheduler, operators must invoke the recovery process; it is not safe to assume background retries exist. + +Apply the generated PostgreSQL schema before enabling this version; it creates the additional cleanup table and due-time index. Attachment ID is optional because rejected uploads may never have a committed ID. The upgrade preserves existing cleanup rows and removes the earlier attachment-ID uniqueness constraint. Do not remove that table during application rollback while cleanup remains pending. Existing releases do not know how to drain it. + +Verification covers real PostgreSQL metadata rejection, outer upload/delete rollback, commit, provider failure followed by a new recovery service, loss of post-commit callbacks, failed queue acknowledgement and repeat replay. Deterministic storage faults also cover a provider that writes bytes before reporting upload failure and two concurrent cleanup workers. A real MinIO fixture, accessed through the published object-storage QBit, verifies deletion replay in plain and versioned buckets while preserving another object's exact bytes. This is not evidence for every object-storage provider. The separate abrupt-JVM test described below uses fixture file storage. + +Failed upload rollback compensation now records its locator independently of the rolled-back transaction before attempting storage cleanup. Failed worker attempts persist a next-attempt time at least 60 seconds later; this lets healthy requests beyond a failing batch proceed. Tests verify later recovery once the failed requests become due. + +Upload I/O and runtime failures attempt durable compensation and preserve the original exception, adding cleanup failure as a suppressed exception. Deferred provider completion is covered by the settlement protocol below. + +Full Maven verification including operator settlement passes 206 tests, zero failures, errors or skips, plus packaging, help coverage, architecture guardrails and JaCoCo checks. PostgreSQL tests cover the retained receipt, repeated requests, active owners, existing references, rejected receipt writes, provider failure and the real form/Review/Result flow with permission revocation. Coordinated host adoption and staging rollback still require separate verification. + +On backends supporting transactional FOR UPDATE SKIP LOCKED (including the PostgreSQL host), upload now commits a cleanup locator before calling storage and holds its row lock for the upload transaction. Metadata insertion and removal of that locator share the same transaction. Recovery workers skip active uploads and hold the same lock while deleting abandoned bytes and acknowledging cleanup. An existing caller transaction retains ownership until it ends; otherwise the service starts its own transaction after spooling the input. Backends without this locking capability retain callback-based compensation and are not covered by this ownership guarantee. + +Focused PostgreSQL tests verify the locator exists before provider I/O, cleanup skips an active upload, and recovery succeeds after an interrupted upload whose rollback callbacks were deliberately suppressed. That test simulates callback loss. A separate child JVM executes the actual upload service against the disposable PostgreSQL database, persists fixture bytes, then calls Runtime.halt before returning from the provider. The parent verifies exit code 73, absent metadata, retained intent and bytes, and successful idempotent recovery using the original locator. This proves abrupt JVM termination recovery, not power-loss durability or late remote-provider completion. Another test rejects intent insertion in PostgreSQL and verifies zero provider writes or deletes; a completed upload in an open outer transaction remains protected from cleanup until commit. Concurrent workers verify that only one worker enters provider deletion for a locator. + +Issue #1289 remains open for coordinated host adoption and application rollback acceptance. Row locking does not make external object storage and the database one atomic transaction. + +## Unreleased late-upload settlement change + +A PostgreSQL regression reproduced a PUT finishing after timeout rollback had deleted the still-empty location and acknowledged its cleanup row. Upload intents (no attachment ID) now use AttachmentStorage.cleanupUpload, which removes current bytes and reports whether an earlier PUT can still complete. The generic adapter cannot prove settlement and returns false. Recovery retains the locator for later sweeps, with the existing 60-second backoff and 100-row batch. Ordinary committed deletion and successful upload acknowledgement remain unchanged. + +Adapters may report settlement only from authoritative completion/cancellation information, never merely object absence. A deterministic deferred-write fixture verifies the late-completion ordering; this does not establish remote S3 settlement. Generic uncertain-upload records remain pending until the provider proves settlement or an administrator performs the explicit operation below. There is no age-based expiry. + +## Resolve a settled upload + +In the Attachment Cleanup table, select upload intents and run **Resolve settled uploads**. An identified user with `collaboration.admin` must supply provider completion or cancellation evidence covering every selected locator, then confirm the standard Review screen. A support case or request receipt establishing completion is suitable; object absence, an elapsed timeout, or an application process stopping does not establish that an external provider has stopped writing. If settlement cannot be established, leave the intent pending. + +Apply rechecks permission, locks each row, rejects active ownership and existing attachment references, and deletes current bytes before saving completion. Provider or database failures leave the intent unresolved for retry. Committed attachment deletions use the ordinary cleanup retry operation instead. + +The service-managed row retains `resolvedAt`, `resolvedBy` and `settlementEvidence`, together with the original locator. Repeated settlement does not overwrite that receipt; scheduled retries exclude resolved rows. Direct edits and deletes remain denied even to administrators. These fields provide the durable operational receipt without relying on the framework's best-effort audit writer. Store a reference to provider evidence, never credentials or private object contents. + +Apply the additive settlement columns and pending-row index before this code runs. Kof22's corresponding migration is `2026-10-26-collaboration-upload-settlement.sql`. Retain receipts and pending intents during rollback. Do not start an earlier cleanup worker that ignores `resolvedAt`: it could consume retained receipts. Roll back to the pre-cleanup application with cleanup processing disabled, or use a version that understands settlement receipts. diff --git a/pom.xml b/pom.xml index 32a601d..47305d1 100644 --- a/pom.xml +++ b/pom.xml @@ -72,6 +72,11 @@ + + com.kingsrook.qqq + qqq-backend-module-postgres + test + com.kingsrook.qbits qbit-redis diff --git a/src/main/java/com/kingsrook/qbits/collaboration/CollaborationQBitProducer.java b/src/main/java/com/kingsrook/qbits/collaboration/CollaborationQBitProducer.java index 5cb6bcf..b389749 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/CollaborationQBitProducer.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/CollaborationQBitProducer.java @@ -187,6 +187,8 @@ public class CollaborationQBitProducer implements QBitMetaDataProducer + { + try + { + QContext.withTemporaryContext(context, () -> finish(pending)); + } + catch(QException e) + { + throw (new IllegalStateException("Attachment cleanup remains durably queued.", e)); + } + }); + return; + } + QRecord pending; + try(QBackendTransaction transaction = QBackendTransaction.openFor(new DeleteInput(CollaborationTables.ATTACHMENT))) + { + QContext.setTransaction(transaction); + try + { + pending = enqueueAndDelete(row, cleanup); + transaction.commit(); + } + catch(QException | RuntimeException failure) + { + transaction.rollback(); + throw (failure); + } + finally + { + QContext.clearTransaction(); + } + } + finish(pending); + } + + + /******************************************************************************* + ** Store the recovery locator in the same database transaction as row deletion. + *******************************************************************************/ + private QRecord enqueueAndDelete(QRecord row, boolean cleanup) throws QException + { + QRecord intent = new QRecord().withValue("attachmentId", row.getValue("id")) + .withValue("bucketName", row.getValue("bucketName")).withValue("storageKey", row.getValue("storageKey")); + InsertInput insert = new InsertInput(CollaborationTables.ATTACHMENT_CLEANUP).withRecords(List.of(intent)) + .withTransaction(QContext.getQBackendTransaction()); + insert.withPermissionBypass("Authorized attachment deletion records durable storage cleanup."); + List inserted = new InsertAction().execute(insert).getRecords(); + AttachmentRecords.checkErrors(inserted); + DeleteInput delete = new DeleteInput(CollaborationTables.ATTACHMENT).withPrimaryKeys(List.of(row.getValue("id"))) + .withTransaction(QContext.getQBackendTransaction()); + delete.setTableMetaData(cleanup ? CollaborationRecords.cleanupMetadata(CollaborationTables.ATTACHMENT) : AttachmentRecords.storageTable()); + delete.withPermissionBypass("Attachment deletion was authorized before recording cleanup."); + AttachmentRecords.checkErrors(new DeleteAction().execute(delete).getRecordsWithErrors()); + return (inserted.getFirst()); + } + + + /******************************************************************************* + ** Durable upload ownership requires real transactional row locking. + *******************************************************************************/ + static boolean supportsUploadOwnership() throws QException + { + var backend = QContext.getQInstance().getBackendForTable(CollaborationTables.ATTACHMENT_CLEANUP); + return (new QBackendModuleDispatcher().getQBackendModule(backend).supportsQueryLockMode(backend, QueryLockMode.FOR_UPDATE_SKIP_LOCKED)); + } + + + /******************************************************************************* + ** Commit the locator independently, then lock it before any provider write. + ** If a worker claimed it first, abort without uploading any bytes. + *******************************************************************************/ + QRecord beginUpload(String bucket, String key) throws QException + { + if(!supportsUploadOwnership()) + { + return (null); + } + QBackendTransaction transaction = QContext.getQBackendTransaction(); + if(transaction == null) + { + throw (new QException("Durable attachment uploads require an active transaction.")); + } + CapturedContext context = new CapturedContext(QContext.getQInstance(), QContext.getQSession()); + QRecord[] pending = new QRecord[1]; + QContext.withTemporaryContext(context, () -> + { + InsertInput insert = new InsertInput(CollaborationTables.ATTACHMENT_CLEANUP).withRecords(List.of( + new QRecord().withValue("bucketName", bucket).withValue("storageKey", key))); + insert.withPermissionBypass("Authorized upload records its immutable locator before external I/O."); + List inserted = new InsertAction().execute(insert).getRecords(); + AttachmentRecords.checkErrors(inserted); + pending[0] = inserted.getFirst(); + }); + if(lock(pending[0], transaction) == null) + { + throw (new QException("Upload cleanup intent was already claimed; retry the upload.")); + } + Runnable cleanup = () -> + { + try + { + QContext.withTemporaryContext(context, () -> finish(pending[0])); + } + catch(QException e) + { + throw (new IllegalStateException("Upload cleanup remains durably queued.", e)); + } + }; + transaction.addAfterRollbackCallback(cleanup); + transaction.addAfterCommitCallback(cleanup); + return (pending[0]); + } + + + /******************************************************************************* + ** Publish metadata and retire its upload intent in the same transaction. + *******************************************************************************/ + void completeUpload(QRecord pending) throws QException + { + if(pending != null) + { + DeleteInput delete = new DeleteInput(CollaborationTables.ATTACHMENT_CLEANUP).withPrimaryKey(pending.getValue("id")) + .withTransaction(QContext.getQBackendTransaction()); + delete.withPermissionBypass("Attachment metadata now owns the uploaded object in this transaction."); + AttachmentRecords.checkErrors(new DeleteAction().execute(delete).getRecordsWithErrors()); + } + } + + + /******************************************************************************* + ** Skip active owners without waiting or updating their locked intent. + *******************************************************************************/ + private QRecord lock(QRecord pending, QBackendTransaction transaction) throws QException + { + QueryInput query = new QueryInput(CollaborationTables.ATTACHMENT_CLEANUP).withTransaction(transaction) + .withLockMode(QueryLockMode.FOR_UPDATE_SKIP_LOCKED) + .withFilter(new QQueryFilter().withCriteria("id", QCriteriaOperator.EQUALS, pending.getValue("id"))); + query.withPermissionBypass("Internal attachment lifecycle owns this immutable cleanup locator."); + List rows = new QueryAction().execute(query).getRecords(); + return (rows.isEmpty() ? null : rows.getFirst()); + } + + + /******************************************************************************* + ** Distinguish an active owner from a committed row requiring delayed retry. + *******************************************************************************/ + private enum CleanupOutcome + { + COMPLETED, ACTIVE, REFERENCED, UNCERTAIN_UPLOAD + } + + + /******************************************************************************* + ** Hold the same row lock as an uploader until deletion and acknowledgement end. + *******************************************************************************/ + private CleanupOutcome finish(QRecord pending) throws QException + { + if(!supportsUploadOwnership()) + { + return (finishUnlocked(pending)); + } + CleanupOutcome[] outcome = { CleanupOutcome.ACTIVE }; + CapturedContext context = new CapturedContext(QContext.getQInstance(), QContext.getQSession()); + QContext.withTemporaryContext(context, () -> + { + try(QBackendTransaction transaction = QBackendTransaction.openFor(new DeleteInput(CollaborationTables.ATTACHMENT_CLEANUP))) + { + QContext.setTransaction(transaction); + try + { + QRecord locked = lock(pending, transaction); + if(locked != null) + { + outcome[0] = finishUnlocked(locked); + } + transaction.commit(); + } + catch(QException | RuntimeException failure) + { + transaction.rollback(); + throw (failure); + } + finally + { + QContext.clearTransaction(); + } + } + }); + return (outcome[0]); + } + + + /******************************************************************************* + ** Never erase bytes while metadata still exists, including nontransactional hosts. + ** Provider or queue deletion failure retains the intent for idempotent replay. + *******************************************************************************/ + private CleanupOutcome finishUnlocked(QRecord pending) throws QException + { + if(pending.getValue("resolvedAt") != null) + { + return (CleanupOutcome.COMPLETED); + } + if(!AttachmentRecords.query(new QQueryFilter() + .withCriteria("bucketName", QCriteriaOperator.EQUALS, pending.getValue("bucketName")) + .withCriteria("storageKey", QCriteriaOperator.EQUALS, pending.getValue("storageKey"))).isEmpty()) + { + return (CleanupOutcome.REFERENCED); + } + if(pending.getValue("attachmentId") == null) + { + if(!storage.cleanupUpload(pending.getValueString("bucketName"), pending.getValueString("storageKey"))) + { + return (CleanupOutcome.UNCERTAIN_UPLOAD); + } + } + else + { + storage.delete(pending.getValueString("bucketName"), pending.getValueString("storageKey")); + } + DeleteInput delete = new DeleteInput(CollaborationTables.ATTACHMENT_CLEANUP).withPrimaryKeys(List.of(pending.getValue("id"))).withTransaction(QContext.getQBackendTransaction()); + delete.withPermissionBypass("Storage cleanup completed; remove its durable intent."); + AttachmentRecords.checkErrors(new DeleteAction().execute(delete).getRecordsWithErrors()); + return (CleanupOutcome.COMPLETED); + } + + + /******************************************************************************* + ** Rollback compensation must survive a provider outage. Persist independently + ** of the failed transaction, then attempt cleanup with the original context. + *******************************************************************************/ + void compensateUpload(String bucket, String key) + { + CapturedContext context = new CapturedContext(QContext.getQInstance(), QContext.getQSession()); + try + { + QContext.withTemporaryContext(context, () -> + { + InsertInput insert = new InsertInput(CollaborationTables.ATTACHMENT_CLEANUP).withRecords(List.of( + new QRecord().withValue("bucketName", bucket).withValue("storageKey", key))); + insert.withPermissionBypass("Failed authorized upload retains its immutable storage locator for compensation."); + List inserted = new InsertAction().execute(insert).getRecords(); + AttachmentRecords.checkErrors(inserted); + finish(inserted.getFirst()); + }); + } + catch(QException e) + { + throw (new IllegalStateException("Upload compensation could not complete.", e)); + } + } + + + /******************************************************************************* + ** Bounded recovery from committed intents, independent of the original service. + *******************************************************************************/ + public int retryPending() throws QException + { + requireOperator(); + if(QContext.getQBackendTransaction() != null) + { + throw (new QException("Attachment cleanup must run outside a caller transaction.")); + } + QueryInput query = new QueryInput(CollaborationTables.ATTACHMENT_CLEANUP).withFilter(new QQueryFilter() + .withCriteria("resolvedAt", QCriteriaOperator.IS_BLANK).withSubFilter(new QQueryFilter() + .withBooleanOperator(QQueryFilter.BooleanOperator.OR) + .withCriteria("nextAttemptAt", QCriteriaOperator.IS_BLANK) + .withCriteria("nextAttemptAt", QCriteriaOperator.LESS_THAN_OR_EQUALS, Instant.now())) + .withOrderBy(new QFilterOrderBy("id")).withLimit(100)); + query.withPermissionBypass("Authorized operator retries committed storage cleanup."); + List pending = new QueryAction().execute(query).getRecords(); + int completed = 0; + QException failure = null; + for(QRecord row : pending) + { + try + { + CleanupOutcome outcome = finish(row); + if(outcome == CleanupOutcome.COMPLETED) + { + completed++; + } + else if(outcome == CleanupOutcome.REFERENCED || outcome == CleanupOutcome.UNCERTAIN_UPLOAD) + { + defer(row); + } + } + catch(QException | RuntimeException e) + { + try + { + defer(row); + } + catch(QException | RuntimeException deferFailure) + { + e.addSuppressed(deferFailure); + } + if(failure == null) + { + failure = new QException("Attachment cleanup remains pending; retry after the provider or database recovers.", e); + } + } + } + if(failure != null) + { + throw (failure); + } + return (completed); + } + + + /******************************************************************************* + ** Persist backoff so failing entries do not monopolize the next bounded batch. + *******************************************************************************/ + private void defer(QRecord pending) throws QException + { + UpdateInput update = new UpdateInput(CollaborationTables.ATTACHMENT_CLEANUP).withRecords(List.of( + new QRecord().withValue("id", pending.getValue("id")).withValue("nextAttemptAt", Instant.now().plusSeconds(60)))); + update.withPermissionBypass("Committed cleanup failed or is not yet safe; retain it for a later attempt."); + AttachmentRecords.checkErrors(new UpdateAction().execute(update).getRecords()); + } + + + /******************************************************************************* + ** An identified administrator attests provider settlement; absence or age alone + ** is not evidence. Keep the receipt on the service-managed row permanently. + *******************************************************************************/ + public void resolveUpload(Serializable id, String evidence) throws QException + { + String actor = requireSettlementOperator(); + validateEvidence(evidence); + if(id == null || QContext.getQBackendTransaction() != null || !supportsUploadOwnership()) + { + throw (new QException("Upload settlement requires a record ID and an independent locking transaction.")); + } + try(QBackendTransaction transaction = QBackendTransaction.openFor(new UpdateInput(CollaborationTables.ATTACHMENT_CLEANUP))) + { + QContext.setTransaction(transaction); + try + { + QRecord pending = lock(new QRecord().withValue("id", id), transaction); + if(pending == null) + { + throw (new QException("Cleanup record is active or no longer exists. Retry after its owner finishes.")); + } + if(pending.getValue("resolvedAt") != null) + { + transaction.commit(); + return; + } + if(pending.getValue("attachmentId") != null) + { + throw (new QException("Use normal cleanup retry for committed attachment deletions.")); + } + if(!AttachmentRecords.query(new QQueryFilter() + .withCriteria("bucketName", QCriteriaOperator.EQUALS, pending.getValue("bucketName")) + .withCriteria("storageKey", QCriteriaOperator.EQUALS, pending.getValue("storageKey"))).isEmpty()) + { + throw (new QException("An attachment still references this object; settlement is not allowed.")); + } + storage.delete(pending.getValueString("bucketName"), pending.getValueString("storageKey")); + UpdateInput update = new UpdateInput(CollaborationTables.ATTACHMENT_CLEANUP).withTransaction(transaction) + .withRecords(List.of(new QRecord().withValue("id", pending.getValue("id")) + .withValue("resolvedAt", Instant.now()).withValue("resolvedBy", actor) + .withValue("settlementEvidence", evidence.strip()))); + update.withPermissionBypass("Identified collaboration administrator records confirmed upload settlement after storage cleanup."); + AttachmentRecords.checkErrors(new UpdateAction().execute(update).getRecords()); + transaction.commit(); + } + catch(QException | RuntimeException failure) + { + transaction.rollback(); + throw (failure); + } + finally + { + QContext.clearTransaction(); + } + } + } + + + /******************************************************************************* + ** The form and service both reject missing or oversized evidence. + *******************************************************************************/ + static void validateEvidence(String evidence) throws QException + { + if(evidence == null || evidence.isBlank() || evidence.strip().length() > 2000) + { + throw (new QException("Record provider completion or cancellation evidence (1–2000 characters).")); + } + } + + + /******************************************************************************* + ** Settlement is a deliberate, attributable admin action, never scheduled work. + *******************************************************************************/ + static String requireSettlementOperator() throws QPermissionDeniedException + { + var session = QContext.getQSession(); + String actor = CollaborationSecurity.userId(session); + if(session == null || session instanceof QSystemUserSession || !session.hasPermission(CollaborationPermissions.ADMIN) + || actor == null || actor.isBlank() || actor.length() > 255) + { + throw (new QPermissionDeniedException("An identified collaboration administrator must confirm upload settlement.")); + } + return (actor); + } + + + /******************************************************************************* + ** Scheduled system work and explicit admin recovery share one authorization rule. + *******************************************************************************/ + static void requireOperator() throws QPermissionDeniedException + { + var session = QContext.getQSession(); + if(!(session instanceof QSystemUserSession) && (session == null || !session.hasPermission(CollaborationPermissions.ADMIN))) + { + throw (new QPermissionDeniedException("This needs the collaboration.admin permission.")); + } + } +} diff --git a/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentStorage.java b/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentStorage.java index 73e7f50..f08ab44 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentStorage.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentStorage.java @@ -73,6 +73,18 @@ public class AttachmentStorage } } + /******************************************************************************* + ** Remove failed-upload bytes and report whether no earlier PUT can still finish. + ** Generic object storage cannot prove remote settlement after a timeout. Keep + ** its durable locator for later sweeps. Adapters may return true only with an + ** authoritative completion/cancellation guarantee, never from object absence. + *******************************************************************************/ + public boolean cleanupUpload(String name, String key) + { + delete(name, key); + return (false); + } + /******************************************************************************* ** Create a temporary link, never a persistent table value. *******************************************************************************/ diff --git a/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsService.java b/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsService.java index fc77167..95294f2 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsService.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsService.java @@ -129,50 +129,11 @@ public class AttachmentsService } } String type = contentType == null || contentType.isBlank() ? "application/octet-stream" : contentType; - var transaction = QContext.getQBackendTransaction(); - if(transaction != null && !transaction.supportsRollbackCallbacks()) - { - throw (new QException("Attachment uploads require a transaction with confirmed rollback support.")); - } - try(InputStream upload = Files.newInputStream(temporary)) - { - storage.put(bucket, PutRequest.of(key, upload).withContentLength(size).withContentType(type).withContentDisposition("inline")); - } - if(transaction != null) - { - var context = new com.kingsrook.qqq.backend.core.context.CapturedContext(QContext.getQInstance(), QContext.getQSession()); - transaction.addAfterRollbackCallback(() -> QContext.withTemporaryContext(context, () -> storage.delete(bucket, key))); - } - boolean rowSaved = false; - try - { - QRecord row = new QRecord().withValue("tableName", tableName).withValue("recordId", id).withValue("fileName", name) - .withValue("contentType", type).withValue("sizeBytes", size).withValue("sha256", HexFormat.of().formatHex(digest.digest())) - .withValue("bucketName", bucket).withValue("storageKey", key).withValue("description", description) - .withValue("createdByUserId", user).withValue("parentDeleted", false); - InsertInput input = new InsertInput(CollaborationTables.ATTACHMENT).withTransaction(QContext.getQBackendTransaction()).withRecords(List.of(row)); - input.withPermissionBypass("AttachmentsService checked parent and attach permission before upload."); - List records = new InsertAction().execute(input).getRecords(); - AttachmentRecords.checkErrors(records); - rowSaved = true; - com.kingsrook.qbits.collaboration.activity.ActivityWriter.write(tableName, id, "FILE_ADDED", "File added", CollaborationTables.ATTACHMENT, records.getFirst().getValueString("id")); - return (entry(records.getFirst())); - } - catch(QException | RuntimeException failure) - { - try - { - if(!rowSaved) - { - storage.delete(bucket, key); - } - } - catch(RuntimeException cleanupFailure) - { - failure.addSuppressed(cleanupFailure); - } - throw (failure); - } + QRecord row = new QRecord().withValue("tableName", tableName).withValue("recordId", id).withValue("fileName", name) + .withValue("contentType", type).withValue("sizeBytes", size).withValue("sha256", HexFormat.of().formatHex(digest.digest())) + .withValue("bucketName", bucket).withValue("storageKey", key).withValue("description", description) + .withValue("createdByUserId", user).withValue("parentDeleted", false); + return (storeUpload(row, temporary)); } catch(IOException | NoSuchAlgorithmException e) { @@ -194,6 +155,107 @@ public class AttachmentsService } } + /******************************************************************************* + ** On transactional locking backends, own a transaction only after spooling + ** the bounded input. An existing outer transaction retains the upload lock. + *******************************************************************************/ + private AttachmentEntry storeUpload(QRecord row, Path temporary) throws QException, IOException + { + var transaction = QContext.getQBackendTransaction(); + if(transaction == null && AttachmentCleanupQueue.supportsUploadOwnership()) + { + try(var owned = com.kingsrook.qqq.backend.core.actions.QBackendTransaction.openFor(new InsertInput(CollaborationTables.ATTACHMENT))) + { + QContext.setTransaction(owned); + try + { + AttachmentEntry result = storeUpload(row, temporary); + owned.commit(); + return (result); + } + catch(QException | IOException | RuntimeException failure) + { + try + { + owned.rollback(); + } + catch(QException rollbackFailure) + { + failure.addSuppressed(rollbackFailure); + } + throw (failure); + } + finally + { + QContext.clearTransaction(); + } + } + } + if(transaction != null && !transaction.supportsRollbackCallbacks()) + { + throw (new QException("Attachment uploads require a transaction with confirmed rollback support.")); + } + String bucket = row.getValueString("bucketName"); + String key = row.getValueString("storageKey"); + AttachmentCleanupQueue cleanup = new AttachmentCleanupQueue(storage); + QRecord pending = cleanup.beginUpload(bucket, key); + try(InputStream upload = Files.newInputStream(temporary)) + { + storage.put(bucket, PutRequest.of(key, upload).withContentLength(row.getValueLong("sizeBytes")) + .withContentType(row.getValueString("contentType")).withContentDisposition("inline")); + } + catch(IOException | RuntimeException failure) + { + if(pending == null) + { + try + { + cleanup.compensateUpload(bucket, key); + } + catch(RuntimeException cleanupFailure) + { + failure.addSuppressed(cleanupFailure); + } + } + throw (failure); + } + if(transaction != null && pending == null) + { + var context = new com.kingsrook.qqq.backend.core.context.CapturedContext(QContext.getQInstance(), QContext.getQSession()); + transaction.addAfterRollbackCallback(() -> QContext.withTemporaryContext(context, + () -> cleanup.compensateUpload(bucket, key))); + } + boolean rowSaved = false; + try + { + InsertInput input = new InsertInput(CollaborationTables.ATTACHMENT).withTransaction(QContext.getQBackendTransaction()).withRecords(List.of(row)); + input.withPermissionBypass("AttachmentsService checked parent and attach permission before upload."); + List records = new InsertAction().execute(input).getRecords(); + AttachmentRecords.checkErrors(records); + rowSaved = true; + cleanup.completeUpload(pending); + com.kingsrook.qbits.collaboration.activity.ActivityWriter.write(row.getValueString("tableName"), row.getValueString("recordId"), + "FILE_ADDED", "File added", CollaborationTables.ATTACHMENT, records.getFirst().getValueString("id")); + return (entry(records.getFirst())); + } + catch(QException | RuntimeException failure) + { + if(!rowSaved && pending == null) + { + try + { + cleanup.compensateUpload(bucket, key); + } + catch(RuntimeException cleanupFailure) + { + failure.addSuppressed(cleanupFailure); + } + } + throw (failure); + } + } + + /******************************************************************************* ** Newest first, independently of collaboration.view. *******************************************************************************/ @@ -301,7 +363,7 @@ public class AttachmentsService } /******************************************************************************* - ** Delete storage first so a provider failure leaves a retryable row. + ** Atomically remove metadata and retain a durable storage cleanup request. *******************************************************************************/ void deleteStoredRow(QRecord row) throws QException { @@ -313,27 +375,7 @@ public class AttachmentsService *******************************************************************************/ void deleteStoredRow(QRecord row, boolean cleanup) throws QException { - if(QContext.getQBackendTransaction() != null) - { - var context = new com.kingsrook.qqq.backend.core.context.CapturedContext(QContext.getQInstance(), QContext.getQSession()); - QContext.getQBackendTransaction().addAfterCommitCallback(() -> - { - try - { - QContext.withTemporaryContext(context, () -> deleteStoredRow(row, cleanup)); - } - catch(QException e) - { - throw (new IllegalStateException("Attachment deletion requires retry", e)); - } - }); - return; - } - storage.delete(row.getValueString("bucketName"), row.getValueString("storageKey")); - DeleteInput input = new DeleteInput(CollaborationTables.ATTACHMENT).withTransaction(QContext.getQBackendTransaction()).withPrimaryKeys(List.of(row.getValue("id"))); - input.setTableMetaData(cleanup ? com.kingsrook.qbits.collaboration.CollaborationRecords.cleanupMetadata(CollaborationTables.ATTACHMENT) : AttachmentRecords.storageTable()); - input.withPermissionBypass("Attachment service authorized deletion, or trusted parent cleanup selected this row."); - AttachmentRecords.checkErrors(new DeleteAction().execute(input).getRecordsWithErrors()); + new AttachmentCleanupQueue(storage).delete(row, cleanup); } /******************************************************************************* diff --git a/src/main/java/com/kingsrook/qbits/collaboration/attachments/ResolveAttachmentUploadProcess.java b/src/main/java/com/kingsrook/qbits/collaboration/attachments/ResolveAttachmentUploadProcess.java new file mode 100644 index 0000000..3106416 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/collaboration/attachments/ResolveAttachmentUploadProcess.java @@ -0,0 +1,227 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * 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 + * + * https://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.collaboration.attachments; + + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; +import com.kingsrook.qbits.collaboration.model.CollaborationTables; +import com.kingsrook.qqq.backend.core.actions.permissions.CustomPermissionChecker; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; +import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput; +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.code.QCodeReference; +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.layout.QIcon; +import com.kingsrook.qqq.backend.core.model.metadata.permissions.MetaDataWithPermissionRules; +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.metadata.processes.QComponentType; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendComponentMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendStepMetaData; +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.ExtractViaQueryStep; +import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface; +import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess; + + +/******************************************************************************* + ** Explicit provider settlement through the standard table action and Review flow. + *******************************************************************************/ +public class ResolveAttachmentUploadProcess implements CustomPermissionChecker +{ + public static final String NAME = "collaborationResolveAttachmentUpload"; + public static final String EVIDENCE = "settlementEvidence"; + + + /******************************************************************************* + ** Administrative table action, never a navigation entry or scheduled process. + *******************************************************************************/ + public static QProcessMetaData metadata() + { + Map values = new HashMap<>(); + values.put(StreamedETLWithFrontendProcess.FIELD_SOURCE_TABLE, CollaborationTables.ATTACHMENT_CLEANUP); + values.put(StreamedETLWithFrontendProcess.FIELD_DESTINATION_TABLE, CollaborationTables.ATTACHMENT_CLEANUP); + values.put(StreamedETLWithFrontendProcess.FIELD_SUPPORTS_FULL_VALIDATION, true); + values.put(StreamedETLWithFrontendProcess.FIELD_DO_FULL_VALIDATION, true); + values.put(StreamedETLWithFrontendProcess.FIELD_PREVIEW_MESSAGE, + "Confirm that the provider has completed or cancelled every selected upload. Object absence or elapsed time alone is not proof. Current bytes will be deleted; the settlement receipt will be retained."); + return (StreamedETLWithFrontendProcess.defineProcessMetaData(ExtractViaQueryStep.class, Review.class, Apply.class, values) + .withName(NAME).withLabel("Resolve settled uploads").withTableName(CollaborationTables.ATTACHMENT_CLEANUP).withIcon(new QIcon("check")) + .withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.NOT_PROTECTED) + .withCustomPermissionChecker(new QCodeReference(ResolveAttachmentUploadProcess.class))) + .withStep(0, new QFrontendStepMetaData().withName("evidence").withLabel("Confirm provider settlement") + .withFormField(new QFieldMetaData(EVIDENCE, QFieldType.STRING).withLabel("Provider completion or cancellation evidence") + .withMaxLength(2000).withIsRequired(true)) + .withComponent(new QFrontendComponentMetaData().withType(QComponentType.EDIT_FORM)))); + } + + + /******************************************************************************* + ** Guard direct process invocation as well as menu visibility. + *******************************************************************************/ + @Override + public void checkPermissionsThrowing(AbstractActionInput input, MetaDataWithPermissionRules metadata) throws QPermissionDeniedException + { + AttachmentCleanupQueue.requireSettlementOperator(); + } + + + /******************************************************************************* + ** Shared standard summary language for Review and Result. + *******************************************************************************/ + private static ProcessSummaryLine line(Status status, String future, String past) + { + ProcessSummaryLine line = new ProcessSummaryLine(status); + line.setSingularFutureMessage(future); + line.setPluralFutureMessage(future); + line.setSingularPastMessage(past); + line.setPluralPastMessage(past); + return (line); + } + + + /******************************************************************************* + ** Review never changes provider state or records a completed settlement. + *******************************************************************************/ + public static class Review extends AbstractTransformStep implements ProcessSummaryProviderInterface + { + private final ProcessSummaryLine selected = line(Status.OK, "uploads will be resolved", "uploads selected"); + private final ProcessSummaryLine skipped = line(Status.WARNING, "resolved records or committed deletions will be skipped", "records skipped"); + + + /******************************************************************************* + ** Require evidence before allowing the administrator to review selected rows. + *******************************************************************************/ + @Override + public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException + { + AttachmentCleanupQueue.requireSettlementOperator(); + AttachmentCleanupQueue.validateEvidence(input.getValueString(EVIDENCE)); + for(QRecord row : input.getRecords()) + { + if(row.getValue("attachmentId") != null || row.getValue("resolvedAt") != null) + { + skipped.incrementCountAndAddPrimaryKey(row.getValue("id")); + } + else + { + output.getRecords().add(row); + selected.incrementCountAndAddPrimaryKey(row.getValue("id")); + } + } + } + + + /******************************************************************************* + ** Keep streamed totals across pages. + *******************************************************************************/ + @Override + public ArrayList getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen) + { + ArrayList lines = new ArrayList<>(); + selected.addSelfToListIfAnyCount(lines); + skipped.addSelfToListIfAnyCount(lines); + return (lines); + } + } + + + /******************************************************************************* + ** Apply checks current authorization, row ownership and references again. + *******************************************************************************/ + public static class Apply extends AbstractLoadStep implements ProcessSummaryProviderInterface + { + private final AttachmentCleanupQueue queue; + private final ProcessSummaryLine done = line(Status.OK, "uploads will be resolved", "uploads resolved"); + private final ProcessSummaryLine failed = line(Status.ERROR, "uploads cannot be resolved", "uploads unresolved; inspect and retry"); + + + /******************************************************************************* + ** Construction is also used during context-free metadata validation. + *******************************************************************************/ + public Apply() + { + this(new AttachmentCleanupQueue()); + } + + + /******************************************************************************* + ** Use the same service with a fixture storage provider in process tests. + *******************************************************************************/ + public Apply(AttachmentCleanupQueue queue) + { + this.queue = queue; + } + + + /******************************************************************************* + ** Each locator has its own transaction and immutable settlement receipt. + *******************************************************************************/ + @Override + public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException + { + AttachmentCleanupQueue.requireSettlementOperator(); + AttachmentCleanupQueue.validateEvidence(input.getValueString(EVIDENCE)); + for(QRecord row : input.getRecords()) + { + try + { + queue.resolveUpload(row.getValue("id"), input.getValueString(EVIDENCE)); + done.incrementCountAndAddPrimaryKey(row.getValue("id")); + } + catch(QException | RuntimeException failure) + { + failed.incrementCountAndAddPrimaryKey(row.getValue("id")); + } + } + } + + + /******************************************************************************* + ** Report actual application results and retain review exclusions. + *******************************************************************************/ + @Override + public ArrayList getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen) + { + ArrayList lines = new ArrayList<>(); + done.addSelfToListIfAnyCount(lines); + failed.addSelfToListIfAnyCount(lines); + if(getTransformStep() instanceof Review review) + { + review.skipped.addSelfToListIfAnyCount(lines); + } + return (lines); + } + } +} diff --git a/src/main/java/com/kingsrook/qbits/collaboration/attachments/RetryAttachmentCleanupProcess.java b/src/main/java/com/kingsrook/qbits/collaboration/attachments/RetryAttachmentCleanupProcess.java new file mode 100644 index 0000000..c48a9bd --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/collaboration/attachments/RetryAttachmentCleanupProcess.java @@ -0,0 +1,99 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * 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 + * + * https://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.collaboration.attachments; + + +import com.kingsrook.qbits.collaboration.CollaborationQBitConfig; +import com.kingsrook.qbits.collaboration.permissions.CollaborationPermissions; +import com.kingsrook.qqq.backend.core.actions.permissions.CustomPermissionChecker; +import com.kingsrook.qqq.backend.core.actions.processes.BackendStep; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; +import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput; +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.metadata.code.QCodeReference; +import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon; +import com.kingsrook.qqq.backend.core.model.metadata.permissions.MetaDataWithPermissionRules; +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.metadata.processes.QBackendStepMetaData; +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.model.session.QSession; +import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession; + + +/******************************************************************************* + ** Scheduled durable attachment cleanup. Runs with its caller's session, which must be a + ** scheduler system session or collaboration admin; never elevates a user. + *******************************************************************************/ +public class RetryAttachmentCleanupProcess implements BackendStep, CustomPermissionChecker +{ + public static final String NAME = "collaborationRetryAttachmentCleanup"; + + + + /******************************************************************************* + ** Always register the backend-only process; scheduling is opt-in. + *******************************************************************************/ + public static QProcessMetaData metadata(CollaborationQBitConfig config) + { + QProcessMetaData process = new QProcessMetaData().withName(NAME).withLabel("Retry attachment cleanup").withIsHidden(true).withIcon(new QIcon("history")) + .withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.HAS_ACCESS_PERMISSION) + .withPermissionBaseName("collaboration").withCustomPermissionChecker(new QCodeReference(RetryAttachmentCleanupProcess.class))) + .withStep(new QBackendStepMetaData().withName("run").withCode(new QCodeReference(RetryAttachmentCleanupProcess.class))); + if(config.getSchedulerName() != null) + { + process.withSchedule(new QScheduleMetaData().withSchedulerName(config.getSchedulerName()).withRepeatSeconds(60).withInitialDelaySeconds(60)); + } + return (process); + } + + + + /******************************************************************************* + ** Hidden is a presentation setting, so explicitly guard execution as well. + *******************************************************************************/ + @Override + public void checkPermissionsThrowing(AbstractActionInput input, MetaDataWithPermissionRules metadata) throws QPermissionDeniedException + { + QSession session = QContext.getQSession(); + if(!(session instanceof QSystemUserSession) && (session == null || !session.hasPermission(CollaborationPermissions.ADMIN))) + { + throw (new QPermissionDeniedException("This needs the collaboration.admin permission.")); + } + } + + + + /******************************************************************************* + ** Retry committed cleanup intents after provider or process recovery. + *******************************************************************************/ + @Override + public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException + { + checkPermissionsThrowing(input, null); + new AttachmentCleanupQueue().retryPending(); + } +} diff --git a/src/main/java/com/kingsrook/qbits/collaboration/model/CollaborationTables.java b/src/main/java/com/kingsrook/qbits/collaboration/model/CollaborationTables.java index 4968f3f..b1ddaf4 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/model/CollaborationTables.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/model/CollaborationTables.java @@ -80,6 +80,7 @@ public final class CollaborationTables public static final String ACTIVITY = "collaboration_activity"; public static final String FOLLOW_DIGEST = "collaboration_follow_digest"; public static final String ATTACHMENT = "collaboration_attachment"; + public static final String ATTACHMENT_CLEANUP = "collaboration_attachment_cleanup"; public static final String FEATURE_SETTING = "collaboration_feature_setting"; public static final String FAVORITE = "collaboration_favorite"; public static final String RECENT = "collaboration_recent"; @@ -202,6 +203,21 @@ public final class CollaborationTables tables.addAll(tagTables(config)); tables.addAll(workTables(config)); tables.addAll(customFieldTables(config)); + tables.add(base(config, ATTACHMENT_CLEANUP, "Attachment Cleanup", "delete") + .withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.NOT_PROTECTED) + .withCustomPermissionChecker(new QCodeReference(com.kingsrook.qbits.collaboration.attachments.AttachmentCleanupQueue.class))) + .withRecordLabelFormatAndFields("Cleanup %s", "id") + .withField(new QFieldMetaData("attachmentId", QFieldType.INTEGER)) + .withField(string("bucketName", "Bucket", 255).withIsRequired(true)) + .withField(string("storageKey", "Storage Key", 1024).withIsRequired(true)) + .withField(new QFieldMetaData("nextAttemptAt", QFieldType.DATE_TIME).withLabel("Next Attempt")) + .withField(new QFieldMetaData("resolvedAt", QFieldType.DATE_TIME).withLabel("Resolved At")) + .withField(string("resolvedBy", "Resolved By", 255)) + .withField(string("settlementEvidence", "Provider Settlement Evidence", 2000)) + .withSection(new QFieldSection("cleanup", new QIcon("delete"), Tier.T1, + List.of("id", "attachmentId", "bucketName", "storageKey", "nextAttemptAt")).withLabel("Cleanup")) + .withSection(new QFieldSection("settlement", new QIcon("check"), Tier.T2, + List.of("resolvedAt", "resolvedBy", "settlementEvidence")).withLabel("Operator Settlement"))); for(QTableMetaData table : tables) { table.withSection(new QFieldSection("dates", new QIcon("calendar_today"), Tier.T3, List.of("createDate", "modifyDate")).withLabel("Dates")); diff --git a/src/main/resources/help/qbit-collaboration.yaml b/src/main/resources/help/qbit-collaboration.yaml index 2b48923..d1e282d 100644 --- a/src/main/resources/help/qbit-collaboration.yaml +++ b/src/main/resources/help/qbit-collaboration.yaml @@ -746,3 +746,31 @@ entries: summary: "Choose the record feature you want to configure." - key: possibleValueSource:collaboration_permission_mode;slot:help summary: "Choose whether a feature accepts editing permission or requires an explicit feature permission." +- key: table:collaboration_attachment_cleanup;slot:help + summary: "Review storage cleanup waiting to finish after an attachment was deleted." + content: | + **How rows get created:** Authorized attachment operations create these records automatically. Deleted files are removed after the database commits. Uncertain uploads remain pending until the provider has finished writing. Ordinary successful cleanup removes its entry; administrator-resolved uploads retain a receipt. Administrators can inspect the records but cannot edit them directly. +- key: table:collaboration_attachment_cleanup;field:attachmentId + summary: "The deleted attachment, when known; failed uploads may have no committed attachment ID." +- key: table:collaboration_attachment_cleanup;field:bucketName + summary: "The storage bucket containing the file awaiting cleanup." +- key: table:collaboration_attachment_cleanup;field:storageKey + summary: "The immutable file location retained until cleanup succeeds." +- key: table:collaboration_attachment_cleanup;field:nextAttemptAt + summary: "Failed cleanup waits at least one minute before another scheduled attempt." +- key: table:collaboration_attachment_cleanup;field:resolvedAt + summary: "When an administrator resolved this upload. Resolved receipts are excluded from retries." +- key: table:collaboration_attachment_cleanup;field:resolvedBy + summary: "The signed-in administrator who resolved this upload. The original identity is retained on repeat requests." +- key: table:collaboration_attachment_cleanup;field:settlementEvidence + summary: "Evidence that the provider completed or cancelled the upload. Never include credentials or private file contents." +- key: table:collaboration_attachment_cleanup;section:settlement + summary: "Retained evidence and identity for an administrator-resolved upload." +- key: process:collaborationResolveAttachmentUpload;slot:help + summary: "Resolve uncertain uploads only after confirming that the provider has finished writing." + content: | + Select upload intents, record evidence of provider completion or cancellation for every selected location, and confirm the Review screen. Object absence or elapsed time alone is not proof. If completion cannot be established, leave the intent pending. The action requires an identified collaboration administrator, refuses active or referenced uploads, deletes current bytes, and retains a settlement receipt. Read Result to see which uploads remain unresolved. Existing attachment deletions use normal cleanup retry instead. +- key: process:collaborationRetryAttachmentCleanup;slot:help + summary: "Retry pending attachment file cleanup after storage or database recovery." + content: | + Processes up to 100 committed cleanup requests. Completed entries are removed; failures remain available for retry. Runs every minute when the host configures a collaboration scheduler. Manual execution requires collaboration administrator access. diff --git a/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTablesGuardrailTest.java b/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTablesGuardrailTest.java index 8b3f910..cc0b464 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTablesGuardrailTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTablesGuardrailTest.java @@ -65,6 +65,7 @@ class CollaborationTablesGuardrailTest extends PersonalServiceFixture Map.entry("collaboration_favorite", "user content: a person stars something"), Map.entry("collaboration_recent", "runtime: a person opens a record"), Map.entry("collaboration_attachment", "user content: a person uploads a file"), + Map.entry("collaboration_attachment_cleanup", "runtime: authorized deletion queues durable storage cleanup"), Map.entry("collaboration_comment", "user content: a person comments"), Map.entry("collaboration_comment_revision", "runtime: a comment is edited"), Map.entry("collaboration_mention", "runtime: a comment mentions someone"), @@ -183,7 +184,7 @@ class CollaborationTablesGuardrailTest extends PersonalServiceFixture private List collaborationTables() { List tables = instance.getTables().values().stream().map(QTableMetaData::getName).filter(n -> n.startsWith("collaboration_")).sorted().toList(); - assertThat(tables).hasSize(19); + assertThat(tables).hasSize(20); return (tables); } diff --git a/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTest.java b/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTest.java index 466cf3c..0057ea6 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/CollaborationTest.java @@ -114,7 +114,7 @@ class CollaborationTest CollaborationTables.SCOPE_TYPE, CollaborationTables.FEATURE, CollaborationTables.PERMISSION_MODE); assertThat(instance.getPossibleValueSource(CollaborationTables.FEATURE).getEnumValues()).hasSize(12); assertThat(instance.getApp("collaboration").getChildren()).extracting(child -> child.getName()) - .containsExactlyInAnyOrder(CollaborationTables.COMMENT, CollaborationTables.ATTACHMENT, CollaborationTables.TAG, CollaborationTables.FIELD_DEFINITION, CollaborationTables.TASK, CollaborationTables.SHARE, CollaborationTables.ACCESS_REQUEST, CollaborationTables.RECORD_ASSIGNMENT, CollaborationTables.ACTIVITY, CollaborationTables.FEATURE_SETTING, CollaborationTables.USER); + .containsExactlyInAnyOrder(CollaborationTables.COMMENT, CollaborationTables.ATTACHMENT, CollaborationTables.ATTACHMENT_CLEANUP, CollaborationTables.TAG, CollaborationTables.FIELD_DEFINITION, CollaborationTables.TASK, CollaborationTables.SHARE, CollaborationTables.ACCESS_REQUEST, CollaborationTables.RECORD_ASSIGNMENT, CollaborationTables.ACTIVITY, CollaborationTables.FEATURE_SETTING, CollaborationTables.USER); assertThat(instance.getApp("collaboration").getWidgets()).containsExactlyElementsOf(CollaborationHome.HOME_WIDGETS); for(String widget : CollaborationHome.HOME_WIDGETS) { @@ -221,7 +221,7 @@ class CollaborationTest QInstance headless = new QInstance(); headless.addBackend(new QBackendMetaData().withName("memory").withBackendType(MemoryBackendModule.class)); new CollaborationQBitProducer().withQBitConfig(config.withAdminApp(false)).produce(headless); - assertThat(headless.getTables()).hasSize(19); + assertThat(headless.getTables()).hasSize(20); assertThat(headless.getApps()).isEmpty(); assertThat(headless.getWidgets()).hasSize(5); } diff --git a/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentTransactionsPostgresTest.java b/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentTransactionsPostgresTest.java new file mode 100644 index 0000000..042f15f --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentTransactionsPostgresTest.java @@ -0,0 +1,1267 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * 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 + * + * https://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.collaboration.attachments; + + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.sql.Connection; +import java.sql.DriverManager; +import java.util.List; +import java.util.Map; +import com.kingsrook.qbits.collaboration.CollaborationSchema; +import com.kingsrook.qbits.collaboration.PersonalServiceFixture; +import com.kingsrook.qbits.objectstorage.api.PutRequest; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput; +import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess; +import com.kingsrook.qqq.backend.module.postgres.model.metadata.PostgreSQLBackendMetaData; +import com.kingsrook.qqq.backend.module.rdbms.actions.RDBMSTransaction; +import com.kingsrook.qqq.backend.module.rdbms.jdbc.ConnectionManager; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +/******************************************************************************* + ** Real PostgreSQL transaction boundaries with deterministic storage fault injection. + *******************************************************************************/ +class AttachmentTransactionsPostgresTest extends PersonalServiceFixture +{ + private static GenericContainer postgres; + private RecordingStorage storage; + private AttachmentsService service; + private static AttachmentCleanupQueue processQueue; + + + /******************************************************************************* + ** An isolated database container is required; never use an application database. + *******************************************************************************/ + @BeforeAll + static void startPostgres() + { + postgres = new GenericContainer<>("postgres:17-alpine") + .withEnv("POSTGRES_PASSWORD", "attachment-fixture-only").withExposedPorts(5432) + .waitingFor(Wait.forLogMessage(".*database system is ready to accept connections.*", 2)); + postgres.start(); + } + + + /******************************************************************************* + ** Stop only the database owned by this fixture. + *******************************************************************************/ + @AfterAll + static void stopPostgres() + { + if(postgres != null) + { + postgres.stop(); + } + } + + + /******************************************************************************* + ** Reuse the authorization fixture, replacing its backend with real PostgreSQL. + *******************************************************************************/ + @Override + @BeforeEach + public void setUp() throws Exception + { + super.setUp(); + instance.getBackends().remove("memory"); + instance.addBackend(new PostgreSQLBackendMetaData().withName("memory").withHostName(postgres.getHost()) + .withPort(postgres.getMappedPort(5432)).withDatabaseName("postgres") + .withUsername("postgres").withPassword("attachment-fixture-only")); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("DROP SCHEMA public CASCADE; CREATE SCHEMA public"); + sql.execute(CollaborationSchema.postgresScript(config)); + sql.execute("CREATE TABLE document (id INTEGER PRIMARY KEY, title TEXT, tenant TEXT)"); + sql.execute("INSERT INTO document VALUES (1, 'Visible', 'a'), (3, 'Private', 'b')"); + } + storage = new RecordingStorage(); + service = new AttachmentsService(config, storage); + } + + + /******************************************************************************* + ** Close owner connections before the isolated schema is replaced. + *******************************************************************************/ + @Override + public void tearDown() + { + ConnectionManager.closeConnectionProviders(instance); + super.tearDown(); + } + + + /******************************************************************************* + ** An independently visible intent precedes external I/O, but active uploads + ** cannot be claimed by a cleanup worker on another database connection. + *******************************************************************************/ + @Test + void uploadIntentPrecedesPutAndProtectsActiveUpload() throws Exception + { + storage.beforePut = () -> + { + try + { + assertThat(pendingCount()).isEqualTo(1); + var context = new com.kingsrook.qqq.backend.core.context.CapturedContext(instance, + new com.kingsrook.qqq.backend.core.model.session.QSystemUserSession()); + QContext.withTemporaryContext(context, () -> assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero()); + assertThat(storage.deleteAttempts).hasValue(0); + } + catch(Exception e) + { + throw (new IllegalStateException(e)); + } + }; + upload(); + assertThat(rowCount()).isEqualTo(1); + assertThat(pendingCount()).isZero(); + assertThat(storage.objects).hasSize(1); + } + + + /******************************************************************************* + ** Rejecting durable intent creation must prevent all provider I/O. + *******************************************************************************/ + @Test + void rejectedUploadIntentPreventsProviderWrite() throws Exception + { + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("CREATE FUNCTION reject_upload_intent() RETURNS trigger LANGUAGE plpgsql AS $$ " + + "BEGIN RAISE EXCEPTION 'injected upload intent rejection'; END $$"); + sql.execute("CREATE TRIGGER reject_upload_intent BEFORE INSERT ON collaboration_attachment_cleanup " + + "FOR EACH ROW EXECUTE FUNCTION reject_upload_intent()"); + } + var puts = new java.util.concurrent.atomic.AtomicInteger(); + storage.beforePut = puts::incrementAndGet; + assertThatThrownBy(this::upload).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class); + assertThat(puts).hasValue(0); + assertThat(storage.objects).isEmpty(); + assertThat(storage.deleteAttempts).hasValue(0); + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isZero(); + assertThat(QContext.getQBackendTransaction()).isNull(); + } + + + /******************************************************************************* + ** Completed provider I/O remains protected until the outer owner commits. + *******************************************************************************/ + @Test + void outerTransactionProtectsUploadedBytesUntilCommit() throws Exception + { + try(var tx = new RDBMSTransaction(connect())) + { + QContext.setTransaction(tx); + upload(); + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isEqualTo(1); + var context = new com.kingsrook.qqq.backend.core.context.CapturedContext(instance, + new com.kingsrook.qqq.backend.core.model.session.QSystemUserSession()); + QContext.withTemporaryContext(context, () -> assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero()); + assertThat(storage.deleteAttempts).hasValue(0); + assertThat(storage.objects).hasSize(1); + tx.commit(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isEqualTo(1); + assertThat(pendingCount()).isZero(); + assertThat(storage.objects).hasSize(1); + } + + + /******************************************************************************* + ** A separate JVM halts after durable provider bytes, without callbacks or finally. + *******************************************************************************/ + @Test + void abruptlyTerminatedJvmLeavesRecoverableUpload(@org.junit.jupiter.api.io.TempDir java.nio.file.Path directory) throws Exception + { + String classpath = System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")); + java.nio.file.Path arguments = directory.resolve("java.args"); + java.nio.file.Files.writeString(arguments, "-cp\n\"" + classpath.replace('\\', '/') + "\"\n" + + UploadCrashProbe.class.getName() + "\n" + postgres.getHost() + "\n" + postgres.getMappedPort(5432) + + "\n\"" + directory.toString().replace('\\', '/') + "\"\n"); + java.nio.file.Path executable = java.nio.file.Path.of(System.getProperty("java.home"), "bin", + System.getProperty("os.name").startsWith("Windows") ? "java.exe" : "java"); + Process child = new ProcessBuilder(executable.toString(), "@" + arguments) + .redirectErrorStream(true).redirectOutput(directory.resolve("child.log").toFile()).start(); + try + { + assertThat(child.waitFor(45, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + assertThat(child.exitValue()).as(java.nio.file.Files.readString(directory.resolve("child.log"))).isEqualTo(73); + } + finally + { + if(child.isAlive()) + { + child.destroyForcibly(); + assertThat(child.waitFor(10, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + } + } + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isEqualTo(1); + java.nio.file.Path object = directory.resolve("uploaded-object"); + assertThat(java.nio.file.Files.readAllBytes(object)).containsExactly(1, 2, 3); + admin(); + assertThat(new AttachmentCleanupQueue(new CrashFileStorage(directory, false)).retryPending()).isEqualTo(1); + assertThat(java.nio.file.Files.exists(object)).isFalse(); + assertThat(pendingCount()).isZero(); + assertThat(new AttachmentCleanupQueue(new CrashFileStorage(directory, false)).retryPending()).isZero(); + } + + + /******************************************************************************* + ** Standalone child uses the real service against the parent's disposable DB. + *******************************************************************************/ + public static class UploadCrashProbe extends PersonalServiceFixture + { + /*************************************************************************** + ** Intentionally never returns: the provider halts this child JVM. + ***************************************************************************/ + public static void main(String[] args) throws Exception + { + var probe = new UploadCrashProbe(); + probe.setUp(); + probe.instance.getBackends().remove("memory"); + probe.instance.addBackend(new PostgreSQLBackendMetaData().withName("memory").withHostName(args[0]) + .withPort(Integer.parseInt(args[1])).withDatabaseName("postgres") + .withUsername("postgres").withPassword("attachment-fixture-only")); + new AttachmentsService(probe.config, new CrashFileStorage(java.nio.file.Path.of(args[2]), true)) + .upload("document", 1, "hello.txt", "text/plain", new ByteArrayInputStream(new byte[] { 1, 2, 3 }), null); + throw (new IllegalStateException("Crash fixture unexpectedly returned")); + } + } + + + /******************************************************************************* + ** One fixture-owned object persists across JVM death independently of SQL. + *******************************************************************************/ + private static class CrashFileStorage extends AttachmentStorage + { + private final java.nio.file.Path directory; + private final boolean haltAfterPut; + + /*************************************************************************** + ** Scope every file to the test runner's disposable directory. + ***************************************************************************/ + private CrashFileStorage(java.nio.file.Path directory, boolean haltAfterPut) + { + this.directory = directory; + this.haltAfterPut = haltAfterPut; + } + + /*************************************************************************** + ** Bound this fixture exactly as the normal injected provider. + ***************************************************************************/ + @Override + public long maxBytes(String name) + { + return (1024L); + } + + /*************************************************************************** + ** Persist bytes and their locator, then terminate without shutdown hooks. + ***************************************************************************/ + @Override + public void put(String name, PutRequest request) + { + try + { + java.nio.file.Files.writeString(directory.resolve("locator"), name + "/" + request.getKey()); + java.nio.file.Files.write(directory.resolve("uploaded-object"), request.getStream().readAllBytes()); + if(haltAfterPut) + { + Runtime.getRuntime().halt(73); + } + } + catch(IOException e) + { + throw (new IllegalStateException(e)); + } + } + + /*************************************************************************** + ** The test runner confirms the only writer JVM has exited before recovery. + ***************************************************************************/ + @Override + public boolean cleanupUpload(String name, String key) + { + delete(name, key); + return (!haltAfterPut); + } + + /*************************************************************************** + ** Verify recovery uses the original locator before deleting fixture bytes. + ***************************************************************************/ + @Override + public void delete(String name, String key) + { + try + { + assertThat(java.nio.file.Files.readString(directory.resolve("locator"))).isEqualTo(name + "/" + key); + java.nio.file.Files.deleteIfExists(directory.resolve("uploaded-object")); + } + catch(IOException e) + { + throw (new IllegalStateException(e)); + } + } + } + + + /******************************************************************************* + ** Losing the upload stack before metadata and rollback callbacks still leaves + ** independently committed intent for a later service instance to recover. + *******************************************************************************/ + @Test + void interruptedUploadSurvivesLostRollbackCallbacks() throws Exception + { + storage.afterPut = () -> { throw new AssertionError("injected interrupted upload"); }; + try(var tx = new RDBMSTransaction(connect()) + { + /*********************************************************************** + ** Simulate loss of process-local callbacks, not a killed-JVM test. + ***********************************************************************/ + @Override + public void addAfterRollbackCallback(Runnable callback) + { + } + }) + { + QContext.setTransaction(tx); + assertThatThrownBy(this::upload).isInstanceOf(AssertionError.class); + tx.rollback(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isEqualTo(1); + assertThat(storage.objects).hasSize(1); + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(storage.objects).isEmpty(); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** Later outer rollback removes both the SQL row and the uploaded object. + *******************************************************************************/ + @Test + void uploadOuterRollbackCompensatesObject() throws Exception + { + try(var tx = new RDBMSTransaction(connect())) + { + QContext.setTransaction(tx); + upload(); + assertThat(storage.objects).hasSize(1); + tx.rollback(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isZero(); + assertThat(storage.objects).isEmpty(); + } + + + /******************************************************************************* + ** A rejected outer transaction cannot delete bytes from an existing attachment. + *******************************************************************************/ + @Test + void deleteOuterRollbackPreservesAttachment() throws Exception + { + AttachmentEntry entry = upload(); + try(var tx = new RDBMSTransaction(connect())) + { + QContext.setTransaction(tx); + service.delete(List.of(entry.id())); + tx.rollback(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isEqualTo(1); + assertThat(storage.objects).hasSize(1); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** A committed deletion removes metadata and bytes after the transaction commits. + *******************************************************************************/ + @Test + void deleteCommitRemovesAttachment() throws Exception + { + AttachmentEntry entry = upload(); + try(var tx = new RDBMSTransaction(connect())) + { + QContext.setTransaction(tx); + service.delete(List.of(entry.id())); + assertThat(storage.objects).hasSize(1); + tx.commit(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isZero(); + assertThat(storage.objects).isEmpty(); + } + + + /******************************************************************************* + ** Database rejection must not leave a visible row whose bytes were already deleted. + *******************************************************************************/ + @Test + void metadataDeleteFailurePreservesDownloadableObject() throws Exception + { + AttachmentEntry entry = upload(); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("CREATE FUNCTION reject_attachment_delete() RETURNS trigger LANGUAGE plpgsql AS $$ " + + "BEGIN RAISE EXCEPTION 'injected metadata delete rejection'; END $$"); + sql.execute("CREATE TRIGGER reject_attachment_delete BEFORE DELETE ON collaboration_attachment " + + "FOR EACH ROW EXECUTE FUNCTION reject_attachment_delete()"); + } + assertThatThrownBy(() -> service.delete(List.of(entry.id()))).isInstanceOf(Exception.class); + assertThat(rowCount()).isEqualTo(1); + assertThat(storage.objects).as("a retained visible attachment must retain its bytes").hasSize(1); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** A provider failure leaves a durable locator; a new worker can retry after restart. + *******************************************************************************/ + @Test + void storageDeleteFailureRetainsIntentAndRetrySucceeds() throws Exception + { + AttachmentEntry entry = upload(); + storage.failDelete = true; + assertThatThrownBy(() -> service.delete(List.of(entry.id()))).isInstanceOf(IllegalStateException.class); + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isEqualTo(1); + assertThat(storage.objects).hasSize(1); + storage.failDelete = false; + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).retryPending()) + .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isZero(); + assertThat(storage.objects).isEmpty(); + } + + + /******************************************************************************* + ** A database commit survives loss of every in-process post-commit callback. + *******************************************************************************/ + @Test + void committedIntentSurvivesLostCallback() throws Exception + { + AttachmentEntry entry = upload(); + try(var tx = new RDBMSTransaction(connect()) + { + /*********************************************************************** + ** Simulate process loss after commit, before any callback can execute. + ***********************************************************************/ + @Override + public void addAfterCommitCallback(Runnable callback) + { + } + }) + { + QContext.setTransaction(tx); + service.delete(List.of(entry.id())); + tx.commit(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isEqualTo(1); + assertThat(storage.objects).hasSize(1); + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(storage.objects).isEmpty(); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** Losing queue acknowledgement after deleting bytes must remain replayable. + *******************************************************************************/ + @Test + void queueAcknowledgementFailureIsReplayable() throws Exception + { + AttachmentEntry entry = upload(); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("CREATE FUNCTION reject_cleanup_ack() RETURNS trigger LANGUAGE plpgsql AS $$ " + + "BEGIN RAISE EXCEPTION 'injected cleanup acknowledgement rejection'; END $$"); + sql.execute("CREATE TRIGGER reject_cleanup_ack BEFORE DELETE ON collaboration_attachment_cleanup " + + "FOR EACH ROW EXECUTE FUNCTION reject_cleanup_ack()"); + } + assertThatThrownBy(() -> service.delete(List.of(entry.id()))).isInstanceOf(Exception.class); + assertThat(rowCount()).isZero(); + assertThat(storage.objects).isEmpty(); + assertThat(pendingCount()).isEqualTo(1); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("DROP TRIGGER reject_cleanup_ack ON collaboration_attachment_cleanup"); + } + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero(); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** Queue observations use independent committed database state. + *******************************************************************************/ + private int pendingCount() throws Exception + { + try(Connection connection = connect(); var sql = connection.createStatement(); + var rows = sql.executeQuery("SELECT COUNT(*) FROM collaboration_attachment_cleanup")) + { + rows.next(); + return (rows.getInt(1)); + } + } + + + /******************************************************************************* + ** Failed first-page entries must not starve later healthy cleanup requests. + *******************************************************************************/ + @Test + void failedBatchDoesNotStarveLaterRequests() throws Exception + { + try(Connection connection = connect(); var sql = connection.prepareStatement( + "INSERT INTO collaboration_attachment_cleanup (attachment_id, bucket_name, storage_key) VALUES (?, 'collaboration', ?)")) + { + for(int id = 1; id <= 101; id++) + { + String key = "queued/" + id; + sql.setInt(1, id); + sql.setString(2, key); + sql.executeUpdate(); + storage.objects.put("collaboration/" + key, new byte[] { 1 }); + if(id <= 100) + { + storage.failedKeys.add(key); + } + } + } + admin(); + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).retryPending()).isInstanceOf(Exception.class); + assertThat(storage.deleteAttempts.get()).isEqualTo(100); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(storage.deleteAttempts.get()).isEqualTo(101); + assertThat(pendingCount()).isEqualTo(100); + assertThat(storage.objects).doesNotContainKey("collaboration/queued/101"); + storage.failedKeys.clear(); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("UPDATE collaboration_attachment_cleanup SET next_attempt_at = CURRENT_TIMESTAMP - INTERVAL '1 second'"); + } + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(100); + assertThat(pendingCount()).isZero(); + assertThat(storage.objects).isEmpty(); + } + + + /******************************************************************************* + ** Failed rollback compensation must preserve a durable locator for later cleanup. + *******************************************************************************/ + @Test + void failedUploadRollbackCompensationRemainsRetryable() throws Exception + { + try(var tx = new RDBMSTransaction(connect())) + { + QContext.setTransaction(tx); + upload(); + storage.failDelete = true; + tx.rollback(); + QContext.clearTransaction(); + } + assertThat(rowCount()).isZero(); + assertThat(storage.objects).hasSize(1); + assertThat(pendingCount()).isEqualTo(1); + storage.failDelete = false; + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(storage.objects).isEmpty(); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** Upgrading the first queue shape preserves pending deletions and is repeatable. + *******************************************************************************/ + @Test + void cleanupSchemaUpgradePreservesPendingRequests() throws Exception + { + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("ALTER TABLE collaboration_attachment_cleanup DROP COLUMN next_attempt_at"); + sql.execute("ALTER TABLE collaboration_attachment_cleanup ALTER COLUMN attachment_id SET NOT NULL"); + sql.execute("CREATE UNIQUE INDEX uq_collaboration_attachment_cleanup_1 ON collaboration_attachment_cleanup (attachment_id)"); + sql.execute("INSERT INTO collaboration_attachment_cleanup (attachment_id, bucket_name, storage_key) VALUES (42, 'collaboration', 'pending/42')"); + sql.execute(CollaborationSchema.postgresScript(config)); + sql.execute(CollaborationSchema.postgresScript(config)); + sql.execute("INSERT INTO collaboration_attachment_cleanup (bucket_name, storage_key) VALUES ('collaboration', 'rollback/one'), ('collaboration', 'rollback/two')"); + } + assertThat(pendingCount()).isEqualTo(3); + } + + + /******************************************************************************* + ** A provider may store bytes and then fail before acknowledging the upload. + *******************************************************************************/ + @Test + void ambiguousUploadFailureRetainsCompensation() throws Exception + { + storage.failPutAfterWrite = true; + storage.failDelete = true; + assertThatThrownBy(this::upload).isInstanceOf(IllegalStateException.class) + .hasMessage("injected upload acknowledgement failure"); + assertThat(rowCount()).isZero(); + assertThat(storage.objects).hasSize(1); + assertThat(pendingCount()).isEqualTo(1); + storage.failDelete = false; + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(storage.objects).isEmpty(); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** Object absence is not proof that a remote upload has stopped. + *******************************************************************************/ + @Test + void genericUploadCleanupDoesNotClaimSettlement() + { + java.util.concurrent.atomic.AtomicInteger deletes = new java.util.concurrent.atomic.AtomicInteger(); + AttachmentStorage uncertain = new AttachmentStorage() + { + @Override + public void delete(String name, String key) + { + deletes.incrementAndGet(); + } + }; + assertThat(uncertain.cleanupUpload("bucket", "key")).isFalse(); + assertThat(deletes.get()).isEqualTo(1); + } + + + /******************************************************************************* + ** Manual settlement retains an attributable receipt and does not retry it. + *******************************************************************************/ + @Test + void operatorSettlementRetainsReceiptAndStopsRetries() throws Exception + { + int id = settlementIntent(); + admin(); + new AttachmentCleanupQueue(storage).resolveUpload(id, " Provider case 123 confirms PUT completion. "); + assertThat(storage.objects).isEmpty(); + try(Connection connection = connect(); var sql = connection.createStatement(); + var row = sql.executeQuery("SELECT resolved_by, settlement_evidence, resolved_at FROM collaboration_attachment_cleanup WHERE id = " + id)) + { + assertThat(row.next()).isTrue(); + assertThat(row.getString(1)).isEqualTo("admin"); + assertThat(row.getString(2)).isEqualTo("Provider case 123 confirms PUT completion."); + assertThat(row.getTimestamp(3)).isNotNull(); + } + int attempts = storage.deleteAttempts.get(); + new AttachmentCleanupQueue(storage).resolveUpload(id, "A repeat must not overwrite the original receipt."); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero(); + assertThat(storage.deleteAttempts.get()).isEqualTo(attempts); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute(CollaborationSchema.postgresScript(config)); + try(var row = sql.executeQuery("SELECT settlement_evidence FROM collaboration_attachment_cleanup WHERE id = " + id)) + { + assertThat(row.next()).isTrue(); + assertThat(row.getString(1)).isEqualTo("Provider case 123 confirms PUT completion."); + } + } + } + + + /******************************************************************************* + ** Deny unauthenticated, non-admin and scheduled callers before provider I/O. + *******************************************************************************/ + @Test + void operatorSettlementRequiresIdentityPermissionAndEvidence() throws Exception + { + int id = settlementIntent(); + AttachmentCleanupQueue queue = new AttachmentCleanupQueue(storage); + assertThatThrownBy(() -> queue.resolveUpload(id, "confirmed")).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + QContext.init(instance, new com.kingsrook.qqq.backend.core.model.session.QSystemUserSession()); + assertThatThrownBy(() -> queue.resolveUpload(id, "confirmed")).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + admin(); + assertThatThrownBy(() -> queue.resolveUpload(id, " ")).hasMessageContaining("evidence"); + assertThatThrownBy(() -> queue.resolveUpload(id, "x".repeat(2001))).hasMessageContaining("evidence"); + assertThatThrownBy(() -> queue.resolveUpload(null, "confirmed")).hasMessageContaining("record ID"); + assertThat(storage.deleteAttempts.get()).isZero(); + assertThat(storage.objects).hasSize(1); + } + + + /******************************************************************************* + ** A locked upload cannot be settled by an administrator on another connection. + *******************************************************************************/ + @Test + void operatorSettlementSkipsActiveOwner() throws Exception + { + int id = settlementIntent(); + admin(); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + connection.setAutoCommit(false); + sql.executeQuery("SELECT id FROM collaboration_attachment_cleanup WHERE id = " + id + " FOR UPDATE").close(); + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")) + .hasMessageContaining("active or no longer exists"); + connection.rollback(); + } + assertThat(storage.deleteAttempts.get()).isZero(); + assertThat(QContext.getQBackendTransaction()).isNull(); + } + + + /******************************************************************************* + ** Even an admin cannot delete a referenced locator or settle normal deletions. + *******************************************************************************/ + @Test + void operatorSettlementProtectsReferencedObjects() throws Exception + { + var attachment = upload(); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.executeUpdate("INSERT INTO collaboration_attachment_cleanup (bucket_name, storage_key) SELECT bucket_name, storage_key FROM collaboration_attachment"); + } + admin(); + int id; + try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery("SELECT id FROM collaboration_attachment_cleanup")) + { + row.next(); + id = row.getInt(1); + } + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")) + .hasMessageContaining("still references"); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.executeUpdate("UPDATE collaboration_attachment_cleanup SET attachment_id = " + attachment.id()); + } + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")) + .hasMessageContaining("normal cleanup retry"); + assertThat(storage.deleteAttempts.get()).isZero(); + assertThat(storage.objects).hasSize(1); + } + + + /******************************************************************************* + ** Provider failure or rejected receipt persistence leaves a retryable locator. + *******************************************************************************/ + @Test + void operatorSettlementFailureRetainsIntent() throws Exception + { + int id = settlementIntent(); + admin(); + storage.failDelete = true; + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")).isInstanceOf(Exception.class); + assertThat(storage.objects).hasSize(1); + storage.failDelete = false; + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("CREATE FUNCTION reject_receipt() RETURNS trigger LANGUAGE plpgsql AS 'BEGIN RAISE EXCEPTION ''receipt failure''; END'; CREATE TRIGGER reject_receipt BEFORE UPDATE ON collaboration_attachment_cleanup FOR EACH ROW EXECUTE FUNCTION reject_receipt()"); + } + assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")).isInstanceOf(Exception.class); + assertThat(storage.objects).isEmpty(); + try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery("SELECT resolved_at FROM collaboration_attachment_cleanup WHERE id = " + id)) + { + assertThat(row.next()).isTrue(); + assertThat(row.getTimestamp(1)).isNull(); + } + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("DROP TRIGGER reject_receipt ON collaboration_attachment_cleanup"); + } + new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed after database recovery"); + assertThat(pendingCount()).isEqualTo(1); + assertThat(QContext.getQBackendTransaction()).isNull(); + } + + + /******************************************************************************* + ** Fixture-only orphan, after its writer has definitively stopped. + *******************************************************************************/ + private int settlementIntent() throws Exception + { + storage.objects.put("fixture/settled", new byte[] { 1, 2, 3 }); + try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery( + "INSERT INTO collaboration_attachment_cleanup (bucket_name, storage_key) VALUES ('fixture', 'settled') RETURNING id")) + { + row.next(); + return (row.getInt(1)); + } + } + + + /******************************************************************************* + ** Genuine form, Review and Result boundaries; revoked access is checked again. + *******************************************************************************/ + @ParameterizedTest + @ValueSource(strings = { "success", "revoked", "storageFailure" }) + @SuppressWarnings("unchecked") + void operatorSettlementProcess(String scenario) throws Exception + { + int id = settlementIntent(); + int deletionId; + try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery( + "INSERT INTO collaboration_attachment_cleanup (attachment_id, bucket_name, storage_key) VALUES (99, 'fixture', 'normal-deletion') RETURNING id")) + { + row.next(); + deletionId = row.getInt(1); + } + var process = instance.getProcess(ResolveAttachmentUploadProcess.NAME); + process.getBackendStep(StreamedETLWithFrontendProcess.STEP_NAME_EXECUTE).getInputMetaData() + .getFieldThrowing(StreamedETLWithFrontendProcess.FIELD_LOAD_CODE).setDefaultValue(new QCodeReference(FixtureSettlement.class)); + processQueue = new AttachmentCleanupQueue(storage); + try + { + RunProcessInput start = new RunProcessInput(); + start.setProcessName(ResolveAttachmentUploadProcess.NAME); + start.addValue("recordsParam", "recordIds"); + start.addValue("recordIds", id + "," + deletionId); + start.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.BREAK); + assertThatThrownBy(() -> new RunProcessAction().execute(start)) + .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + admin(); + var form = new RunProcessAction().execute(start); + assertThat(form.getException()).isEmpty(); + start.setProcessUUID(form.getProcessUUID()); + start.setStartAfterStep("evidence"); + start.addValue(ResolveAttachmentUploadProcess.EVIDENCE, "Provider request 123 completed; writer stopped."); + var review = new RunProcessAction().execute(start); + assertThat(review.getException()).isEmpty(); + List reviewLines = (List) review.getValues().get(StreamedETLWithFrontendProcess.FIELD_VALIDATION_SUMMARY); + assertThat(reviewLines).hasSize(2); + assertThat(storage.deleteAttempts.get()).isZero(); + if(scenario.equals("revoked")) + { + login("admin"); + } + storage.failDelete = scenario.equals("storageFailure"); + RunProcessInput go = new RunProcessInput(); + go.setProcessName(ResolveAttachmentUploadProcess.NAME); + go.setProcessUUID(review.getProcessUUID()); + go.setStartAfterStep(StreamedETLWithFrontendProcess.STEP_NAME_REVIEW); + go.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.SKIP); + if(scenario.equals("revoked")) + { + assertThatThrownBy(() -> new RunProcessAction().execute(go)) + .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + assertThat(storage.deleteAttempts.get()).isZero(); + } + else + { + var result = new RunProcessAction().execute(go); + assertThat(result.getException()).isEmpty(); + List lines = (List) result.getValues().get(StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY); + assertThat(lines).anySatisfy(line -> + { + line.pickMessage(true); + assertThat(line.getMessage()).isEqualTo(scenario.equals("success") ? "uploads resolved" : "uploads unresolved; inspect and retry"); + assertThat(line.getCount()).isEqualTo(1); + }); + } + assertThat(storage.objects).hasSize(scenario.equals("success") ? 0 : 1); + } + finally + { + processQueue = null; + } + } + + + /******************************************************************************* + ** The real process uses the real queue with deterministic fixture storage. + *******************************************************************************/ + public static class FixtureSettlement extends ResolveAttachmentUploadProcess.Apply + { + /******************************************************************************* + ** Instantiated by the standard process loader. + *******************************************************************************/ + public FixtureSettlement() + { + super(processQueue); + } + } + + + /******************************************************************************* + ** A timed-out remote PUT can finish after rollback has already attempted DELETE. + *******************************************************************************/ + @Test + void lateUploadCompletionRemainsRecoverable() throws Exception + { + storage.deferPutUntilAfterFailure = true; + assertThatThrownBy(this::upload).isInstanceOf(IllegalStateException.class) + .hasMessage("injected upload timeout"); + assertThat(rowCount()).isZero(); + assertThat(storage.objects).isEmpty(); + + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero(); + assertThat(pendingCount()).isEqualTo(1); + int attempts = storage.deleteAttempts.get(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero(); + assertThat(storage.deleteAttempts.get()).as("uncertain writes use persisted backoff").isEqualTo(attempts); + + // The remote service finishes after rollback and an independent cleanup sweep. + storage.deferredPut.run(); + assertThat(storage.objects).hasSize(1); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.executeUpdate("UPDATE collaboration_attachment_cleanup SET next_attempt_at = CURRENT_TIMESTAMP - INTERVAL '1 second'"); + } + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + assertThat(storage.objects).as("late provider bytes must remain discoverable for cleanup").isEmpty(); + assertThat(pendingCount()).isZero(); + } + + + /******************************************************************************* + ** Replay real provider deletes after the database rejects queue acknowledgement. + *******************************************************************************/ + @Test + void minioDeletionReplayPreservesOtherObjects() throws Exception + { + try(var minio = new GenericContainer<>("cgr.dev/chainguard/minio@sha256:dc455cdd4db6e390b1b26514f15ff37681a777bf8dab41aac445a0259f751e56") + .withEnv("MINIO_ROOT_USER", "collaboration-test").withEnv("MINIO_ROOT_PASSWORD", "collaboration-test-password") + .withCommand("server", "/tmp/minio-data").withExposedPorts(9000) + .waitingFor(Wait.forHttp("/minio/health/live").forPort(9000))) + { + minio.start(); + String endpoint = "http://" + minio.getHost() + ":" + minio.getMappedPort(9000); + try(var client = software.amazon.awssdk.services.s3.S3Client.builder() + .endpointOverride(java.net.URI.create(endpoint)).region(software.amazon.awssdk.regions.Region.US_EAST_1) + .forcePathStyle(true).credentialsProvider(software.amazon.awssdk.auth.credentials.StaticCredentialsProvider.create( + software.amazon.awssdk.auth.credentials.AwsBasicCredentials.create("collaboration-test", "collaboration-test-password"))).build()) + { + client.createBucket(b -> b.bucket("collaboration-plain")); + client.createBucket(b -> b.bucket("collaboration-versioned")); + client.putBucketVersioning(b -> b.bucket("collaboration-versioned").versioningConfiguration(v -> + v.status(software.amazon.awssdk.services.s3.model.BucketVersioningStatus.ENABLED))); + System.setProperty("collaboration.minio.access", "collaboration-test"); + System.setProperty("collaboration.minio.secret", "collaboration-test-password"); + var storageConfig = new com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig() + .withBackendName("memory") + .withAllowedSecretRefPatterns(List.of("prop.collaboration.minio.*")) + .withBucketScoping(false) + .withConnection(new com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig("fixture") + .withEndpoint(endpoint).withRegion("us-east-1") + .withAddressingStyle(com.kingsrook.qbits.objectstorage.config.AddressingStyle.PATH) + .withChecksumMode(com.kingsrook.qbits.objectstorage.config.ChecksumMode.WHEN_REQUIRED) + .withAccessKeyRef("${prop.collaboration.minio.access}").withSecretKeyRef("${prop.collaboration.minio.secret}")); + for(String alias : List.of("plain", "versioned")) + { + storageConfig.withBucket(new com.kingsrook.qbits.objectstorage.config.StorageBucketConfig(alias) + .withConnection("fixture").withBucketName("collaboration-" + alias) + .withRole(com.kingsrook.qbits.objectstorage.config.BucketRole.PRIVATE).withIndexed(false)); + } + com.kingsrook.qbits.objectstorage.api.ObjectStorage.configureStandalone(storageConfig); + for(String alias : List.of("plain", "versioned")) + { + config.withAttachmentsBucketName(alias); + service = new AttachmentsService(config, new AttachmentStorage()); + AttachmentEntry removed = upload(); + AttachmentEntry retained = upload(); + String removedKey = storedKey(removed.id()); + String retainedKey = storedKey(retained.id()); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("CREATE OR REPLACE FUNCTION reject_provider_ack() RETURNS trigger LANGUAGE plpgsql AS $$ " + + "BEGIN RAISE EXCEPTION 'injected provider acknowledgement rejection'; END $$"); + sql.execute("CREATE TRIGGER reject_provider_ack BEFORE DELETE ON collaboration_attachment_cleanup " + + "FOR EACH ROW EXECUTE FUNCTION reject_provider_ack()"); + } + assertThatThrownBy(() -> service.delete(List.of(removed.id()))).isInstanceOf(Exception.class); + assertThat(pendingCount()).isEqualTo(1); + try(Connection connection = connect(); var sql = connection.createStatement()) + { + sql.execute("DROP TRIGGER reject_provider_ack ON collaboration_attachment_cleanup"); + } + admin(); + assertThat(new AttachmentCleanupQueue(new AttachmentStorage()).retryPending()).isEqualTo(1); + assertThat(pendingCount()).isZero(); + var bucket = com.kingsrook.qbits.objectstorage.api.ObjectStorage.asSystem("collaboration-test", "replay").bucket(alias); + assertThat(bucket.head(removedKey)).isEmpty(); + assertThat(bucket.head(retainedKey)).isPresent(); + assertThat(client.getObjectAsBytes(b -> b.bucket("collaboration-" + alias).key(retainedKey)).asByteArray()) + .containsExactly((byte) 1, (byte) 2, (byte) 3); + service.delete(List.of(retained.id())); + login("alice"); + } + } + } + finally + { + com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime.reset(); + com.kingsrook.qbits.objectstorage.internal.ConfigSync.reset(); + System.clearProperty("collaboration.minio.access"); + System.clearProperty("collaboration.minio.secret"); + } + } + + + /******************************************************************************* + ** Observe the immutable provider key from committed SQL metadata. + *******************************************************************************/ + private String storedKey(int attachmentId) throws Exception + { + try(Connection connection = connect(); var sql = connection.prepareStatement("SELECT storage_key FROM collaboration_attachment WHERE id = ?")) + { + sql.setInt(1, attachmentId); + try(var rows = sql.executeQuery()) + { + assertThat(rows.next()).isTrue(); + return (rows.getString(1)); + } + } + } + + + /******************************************************************************* + ** Two workers may observe the same committed intent; replay must remain safe. + *******************************************************************************/ + @Test + void concurrentWorkersCanReplayTheSameIntent() throws Exception + { + AttachmentEntry entry = upload(); + storage.failDelete = true; + assertThatThrownBy(() -> service.delete(List.of(entry.id()))).isInstanceOf(IllegalStateException.class); + storage.failDelete = false; + storage.concurrentDeletes = new java.util.concurrent.CountDownLatch(1); + storage.allowDelete = new java.util.concurrent.CountDownLatch(1); + admin(); + var session = QContext.getQSession(); + try(var executor = java.util.concurrent.Executors.newFixedThreadPool(2)) + { + java.util.concurrent.Callable retry = () -> + { + QContext.init(instance, session); + try + { + return (new AttachmentCleanupQueue(storage).retryPending()); + } + finally + { + QContext.clear(); + } + }; + var first = executor.submit(retry); + assertThat(storage.concurrentDeletes.await(10, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + try + { + var second = executor.submit(retry); + assertThat(second.get(10, java.util.concurrent.TimeUnit.SECONDS)).isZero(); + } + finally + { + storage.allowDelete.countDown(); + } + assertThat(first.get(20, java.util.concurrent.TimeUnit.SECONDS)).isEqualTo(1); + } + assertThat(rowCount()).isZero(); + assertThat(pendingCount()).isZero(); + assertThat(storage.objects).isEmpty(); + } + + + /******************************************************************************* + ** Recovery locations are admin-only and cannot be forged through table writes. + *******************************************************************************/ + @Test + void cleanupQueueIsServiceManaged() throws Exception + { + var query = new com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput( + com.kingsrook.qbits.collaboration.model.CollaborationTables.ATTACHMENT_CLEANUP); + assertThatThrownBy(() -> new com.kingsrook.qqq.backend.core.actions.tables.QueryAction().execute(query)) + .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + admin(); + assertThat(new com.kingsrook.qqq.backend.core.actions.tables.QueryAction().execute(query).getRecords()).isEmpty(); + var insert = new com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput( + com.kingsrook.qbits.collaboration.model.CollaborationTables.ATTACHMENT_CLEANUP) + .withRecords(List.of(new com.kingsrook.qqq.backend.core.model.data.QRecord())); + assertThatThrownBy(() -> new com.kingsrook.qqq.backend.core.actions.tables.InsertAction().execute(insert)) + .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class); + } + + + /******************************************************************************* + ** Use the actual service and its caller's record permissions. + *******************************************************************************/ + private AttachmentEntry upload() throws Exception + { + return (service.upload("document", 1, "hello.txt", "text/plain", new ByteArrayInputStream(new byte[] { 1, 2, 3 }), null)); + } + + + /******************************************************************************* + ** Independent JDBC observation, outside the service transaction. + *******************************************************************************/ + private int rowCount() throws Exception + { + try(Connection connection = connect(); var sql = connection.createStatement(); + var rows = sql.executeQuery("SELECT COUNT(*) FROM collaboration_attachment")) + { + rows.next(); + return (rows.getInt(1)); + } + } + + + /******************************************************************************* + ** Connect only to this disposable fixture's mapped port. + *******************************************************************************/ + private static Connection connect() throws Exception + { + return (DriverManager.getConnection("jdbc:postgresql://" + postgres.getHost() + ":" + + postgres.getMappedPort(5432) + "/postgres", "postgres", "attachment-fixture-only")); + } + + + /******************************************************************************* + ** Preserve bytes and inject provider failure independently of PostgreSQL. + *******************************************************************************/ + private static class RecordingStorage extends AttachmentStorage + { + private final Map objects = new java.util.concurrent.ConcurrentHashMap<>(); + private boolean failDelete; + private boolean failPutAfterWrite; + private boolean deferPutUntilAfterFailure; + private Runnable deferredPut; + private Runnable beforePut; + private Runnable afterPut; + private final java.util.Set failedKeys = new java.util.HashSet<>(); + private final java.util.concurrent.atomic.AtomicInteger deleteAttempts = new java.util.concurrent.atomic.AtomicInteger(); + private java.util.concurrent.CountDownLatch concurrentDeletes; + private java.util.concurrent.CountDownLatch allowDelete; + + /*************************************************************************** + ** Keep the production service's upload bound active. + ***************************************************************************/ + @Override + public long maxBytes(String name) + { + return (1024L); + } + + /*************************************************************************** + ** Consume the same upload stream a real provider receives. + ***************************************************************************/ + @Override + public void put(String name, PutRequest request) + { + try + { + if(beforePut != null) + { + beforePut.run(); + } + byte[] bytes = request.getStream().readAllBytes(); + if(deferPutUntilAfterFailure) + { + deferredPut = () -> + { + objects.put(name + "/" + request.getKey(), bytes); + deferPutUntilAfterFailure = false; + }; + throw (new IllegalStateException("injected upload timeout")); + } + objects.put(name + "/" + request.getKey(), bytes); + if(afterPut != null) + { + afterPut.run(); + } + if(failPutAfterWrite) + { + throw (new IllegalStateException("injected upload acknowledgement failure")); + } + } + catch(IOException e) + { + throw (new IllegalStateException(e)); + } + } + + /*************************************************************************** + ** This synchronous fixture knows whether its deferred PUT has completed. + ***************************************************************************/ + @Override + public boolean cleanupUpload(String name, String key) + { + delete(name, key); + return (!deferPutUntilAfterFailure); + } + + /*************************************************************************** + ** Simulate a provider refusing deletion without losing stored bytes. + ***************************************************************************/ + @Override + public void delete(String name, String key) + { + deleteAttempts.incrementAndGet(); + if(concurrentDeletes != null) + { + concurrentDeletes.countDown(); + try + { + if(!allowDelete.await(20, java.util.concurrent.TimeUnit.SECONDS)) + { + throw (new IllegalStateException("Timed out waiting to release the cleanup owner")); + } + } + catch(InterruptedException e) + { + Thread.currentThread().interrupt(); + throw (new IllegalStateException(e)); + } + } + if(failDelete || failedKeys.contains(key)) + { + throw (new IllegalStateException("injected storage delete failure")); + } + objects.remove(name + "/" + key); + } + } +} diff --git a/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsServiceTest.java b/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsServiceTest.java index 1b0d7f1..d8b4dea 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsServiceTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentsServiceTest.java @@ -383,7 +383,7 @@ public class AttachmentsServiceTest extends PersonalServiceFixture } /******************************************************************************* - ** Storage failures leave rows for retries; trusted cleanup works after parent deletion. + ** Storage failures leave durable cleanup; trusted cleanup works after parent deletion. *******************************************************************************/ @Test void failuresAndCleanup() throws Exception @@ -391,8 +391,13 @@ public class AttachmentsServiceTest extends PersonalServiceFixture AttachmentEntry row = upload("file.txt", "x"); storage.failDelete = true; assertThatThrownBy(() -> service.delete(List.of(row.id()))).hasMessage("Storage unavailable"); - assertThat(service.list("document", 1)).hasSize(1); + assertThat(service.list("document", 1)).isEmpty(); + assertThat(storage.objects).hasSize(1); storage.failDelete = false; + admin(); + assertThat(new AttachmentCleanupQueue(storage).retryPending()).isEqualTo(1); + login("alice"); + upload("parent-cleanup.txt", "parent bytes"); new DeleteAction().execute(new DeleteInput("document").withPrimaryKeys(List.of(1))); assertThatThrownBy(() -> service.list("document", 1)).hasMessage("Not found"); QContext.getQSession().getPermissions().clear(); @@ -400,6 +405,8 @@ public class AttachmentsServiceTest extends PersonalServiceFixture cleanup.onParentsDeleted("document", List.of()); cleanup.onParentsDeleted("document", List.of(1)); cleanup.onParentsDeleted("document", List.of(1)); + admin(); + new AttachmentCleanupQueue(storage).retryPending(); assertThat(storage.objects).isEmpty(); assertThat(AttachmentRecords.query(new QQueryFilter())).isEmpty(); }