From d5d5bf7a960a6ae7161841a3e7b93758f602db4c Mon Sep 17 00:00:00 2001 From: Leo Galambos Date: Mon, 3 Aug 2026 02:26:42 +0200 Subject: [PATCH] feat(pki): add durable publication lifecycle Add durable post-commit publication registration, dispatch, retry and reconciliation over transactional metadata. Keep credentials and status objects authoritative independently of publication outcomes while safely preserving unknown external results. --- .../zeroecho/pki/api/PublicationService.java | 87 ++- .../api/publication/PublicationCursor.java | 30 + .../api/publication/PublicationEvidence.java | 13 + .../api/publication/PublicationFailure.java | 19 + .../pki/api/publication/PublicationQuery.java | 9 +- .../api/publication/PublicationRecord.java | 101 ++- .../api/publication/PublicationRequest.java | 27 + .../api/publication/PublicationResult.java | 11 +- .../publication/PublicationSourceType.java | 13 + .../api/publication/PublicationStatus.java | 24 +- .../api/publication/PublicationTarget.java | 21 +- .../pki/api/publication/package-info.java | 14 +- .../impl/core/DefaultPublicationService.java | 445 +++++++++++++ .../pki/impl/fs/FilesystemPkiStore.java | 335 +++++++++- .../java/zeroecho/pki/impl/fs/FsCodec.java | 61 +- .../java/zeroecho/pki/impl/fs/FsPaths.java | 9 - .../pki/impl/fs/FsSnapshotExporter.java | 66 +- .../pki/impl/fs/PublicationRecordCodec.java | 305 +++++++++ .../pki/spi/publish/PublicationAttempt.java | 43 ++ .../pki/spi/publish/PublicationOutcome.java | 15 + .../zeroecho/pki/spi/publish/Publisher.java | 34 +- .../pki/spi/publish/package-info.java | 4 +- .../java/zeroecho/pki/spi/store/PkiStore.java | 58 +- .../pki/spi/store/PkiStoreProvider.java | 3 + .../zeroecho/pki/e2e/PkiProofGateE2eTest.java | 5 +- .../fs/DefaultPublicationServiceTest.java | 624 ++++++++++++++++++ .../pki/impl/fs/FilesystemPkiStoreTest.java | 17 +- 27 files changed, 2161 insertions(+), 232 deletions(-) create mode 100644 pki/src/main/java/zeroecho/pki/api/publication/PublicationCursor.java create mode 100644 pki/src/main/java/zeroecho/pki/api/publication/PublicationEvidence.java create mode 100644 pki/src/main/java/zeroecho/pki/api/publication/PublicationFailure.java create mode 100644 pki/src/main/java/zeroecho/pki/api/publication/PublicationRequest.java create mode 100644 pki/src/main/java/zeroecho/pki/api/publication/PublicationSourceType.java create mode 100644 pki/src/main/java/zeroecho/pki/impl/core/DefaultPublicationService.java create mode 100644 pki/src/main/java/zeroecho/pki/impl/fs/PublicationRecordCodec.java create mode 100644 pki/src/main/java/zeroecho/pki/spi/publish/PublicationAttempt.java create mode 100644 pki/src/main/java/zeroecho/pki/spi/publish/PublicationOutcome.java create mode 100644 pki/src/test/java/zeroecho/pki/impl/fs/DefaultPublicationServiceTest.java diff --git a/pki/src/main/java/zeroecho/pki/api/PublicationService.java b/pki/src/main/java/zeroecho/pki/api/PublicationService.java index cc47b90..1c4fdab 100644 --- a/pki/src/main/java/zeroecho/pki/api/PublicationService.java +++ b/pki/src/main/java/zeroecho/pki/api/PublicationService.java @@ -33,66 +33,85 @@ ******************************************************************************/ package zeroecho.pki.api; -import java.util.List; +import java.util.Optional; +import zeroecho.pki.api.publication.PublicationCursor; import zeroecho.pki.api.publication.PublicationQuery; import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationRequest; import zeroecho.pki.api.publication.PublicationResult; -import zeroecho.pki.api.publication.PublicationTarget; /** * Publication and distribution operations. * *

* Publishing is an explicit operation enabling parity with established PKI - * systems. Implementations may publish credentials, CA materials, and status - * objects to configured targets such as filesystem mirrors, LDAP directories, - * HTTP endpoints, or object stores. + * systems. Implementations publish committed credential and status-object + * content to administrator-configured targets such as filesystem mirrors, LDAP + * directories, HTTP endpoints, or object stores. Publication never establishes + * source-object authority. *

*/ -public interface PublicationService { +public interface PublicationService extends AutoCloseable { /** - * Publishes an issued credential to the specified target. + * Registers one publication operation after its source object is committed. * - * @param credentialId credential id - * @param target publication target - * @return publication result - * @throws IllegalArgumentException if inputs are invalid - * @throws PkiException if publication fails + * @param request immutable registration request + * @return committed pending record, or the identical existing operation + * @throws IllegalArgumentException if the request is invalid or conflicting + * @throws PkiException if registration cannot be established durably */ - PublicationResult publishCredential(PkiId credentialId, PublicationTarget target); + PublicationRecord register(PublicationRequest request); /** - * Publishes CA materials (e.g., CA certificate sets) for the given CA entity to - * the specified target. + * Processes one pending operation synchronously. * - * @param caId CA entity id - * @param target publication target - * @return publication result - * @throws IllegalArgumentException if inputs are invalid - * @throws PkiException if publication fails + * @param publicationId operation identity + * @return durably classified result + * @throws IllegalArgumentException if the operation is absent or ineligible + * @throws PkiException if persistence or content validation fails */ - PublicationResult publishCaMaterials(PkiId caId, PublicationTarget target); + PublicationResult process(PkiId publicationId); /** - * Publishes a status object to the specified target. + * Explicitly retries one retryable operation as a new numbered attempt. * - * @param statusObjectId status object id - * @param target publication target - * @return publication result - * @throws IllegalArgumentException if inputs are invalid - * @throws PkiException if publication fails + * @param publicationId operation identity + * @return durably classified result + * @throws IllegalArgumentException if the operation is not retryable + * @throws PkiException if persistence or content validation fails */ - PublicationResult publishStatusObject(PkiId statusObjectId, PublicationTarget target); + PublicationResult retry(PkiId publicationId); /** - * Lists publication records matching query constraints. + * Reconciles one unknown external outcome using the original attempt token. * - * @param query publication query - * @return publication records - * @throws IllegalArgumentException if {@code query} is invalid - * @throws PkiException if listing fails + * @param publicationId operation identity + * @return current durable result; unresolved operations remain unknown + * @throws IllegalArgumentException if the operation is not unknown + * @throws PkiException if reconciliation persistence fails */ - List listPublications(PublicationQuery query); + PublicationResult reconcile(PkiId publicationId); + + /** + * Reads one durable publication operation. + * + * @param publicationId operation identity + * @return current record, or empty when absent + */ + Optional find(PkiId publicationId); + + /** + * Opens a stable lazy query over publication operations. + * + * @param query finite query constraints + * @return closeable cursor + * @throws PkiException if the stable view cannot be opened + */ + PublicationCursor openPublications(PublicationQuery query); + + /** Rejects new operations while allowing an in-flight synchronous call to classify its result. */ + @Override + void close(); } diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationCursor.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationCursor.java new file mode 100644 index 0000000..d54e27d --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationCursor.java @@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.api.publication; + +import java.io.IOException; +import java.util.Optional; +import zeroecho.core.io.CancellationSignal; + +/** + * Closeable snapshot-consistent cursor over publication records. + * + *

Records are returned in canonical publication-identity order. Cursor heap + * does not grow with the number of returned records.

+ */ +public interface PublicationCursor extends AutoCloseable { + /** + * Advances by at most one matching record. + * + * @param cancellation cooperative cancellation signal + * @return next record, or empty at end of input + * @throws IOException if the stable metadata view cannot be read + */ + Optional next(CancellationSignal cancellation) throws IOException; + + /** Closes the underlying stable metadata view idempotently. */ + @Override + void close(); +} diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationEvidence.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationEvidence.java new file mode 100644 index 0000000..bdc51cf --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationEvidence.java @@ -0,0 +1,13 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.api.publication; + +/** Identifies how an exact attempt outcome was established. */ +public enum PublicationEvidence { + /** The publisher returned a definitive result from direct dispatch. */ + DIRECT_RESPONSE, + /** The publisher resolved the existing attempt through reconciliation. */ + RECONCILIATION_RESPONSE +} diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationFailure.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationFailure.java new file mode 100644 index 0000000..5f83556 --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationFailure.java @@ -0,0 +1,19 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.api.publication; + +/** Safe closed classifications for publication-operation failures. */ +public enum PublicationFailure { + /** The publisher authoritatively classified the attempt as retryable. */ + EXTERNAL_RETRYABLE, + /** The publisher authoritatively classified the attempt as terminal. */ + EXTERNAL_TERMINAL, + /** The publisher call did not establish whether an external effect occurred. */ + EXTERNAL_OUTCOME_UNKNOWN, + /** The exact immutable source content could not be validated. */ + CONTENT_INVALID, + /** Durable persistence of a returned result could not be established. */ + RESULT_PERSISTENCE_UNKNOWN +} diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationQuery.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationQuery.java index f68ee46..20f9383 100644 --- a/pki/src/main/java/zeroecho/pki/api/publication/PublicationQuery.java +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationQuery.java @@ -42,10 +42,10 @@ import java.util.Optional; * @param targetType optional target type filter * @param after optional lower bound for time * @param before optional upper bound for time - * @param objectKind optional object kind filter + * @param sourceType optional source-object type filter */ public record PublicationQuery(Optional targetType, Optional after, - Optional before, Optional objectKind) { + Optional before, Optional sourceType) { /** * Creates a publication query. @@ -53,8 +53,11 @@ public record PublicationQuery(Optional targetType, Optio * @throws IllegalArgumentException if any optional container is null */ public PublicationQuery { - if (targetType == null || after == null || before == null || objectKind == null) { + if (targetType == null || after == null || before == null || sourceType == null) { throw new IllegalArgumentException("optional fields must not be null"); } + if (after.isPresent() && before.isPresent() && after.orElseThrow().isAfter(before.orElseThrow())) { + throw new IllegalArgumentException("after must not be later than before"); + } } } diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationRecord.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationRecord.java index 48d1437..746dc40 100644 --- a/pki/src/main/java/zeroecho/pki/api/publication/PublicationRecord.java +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationRecord.java @@ -34,11 +34,14 @@ package zeroecho.pki.api.publication; import java.time.Instant; +import java.util.Objects; +import java.util.Optional; import zeroecho.pki.api.PkiId; +import zeroecho.pki.api.content.DurableContentReference; /** - * Persisted record of a publication attempt. + * Persisted current state of one publication operation. * *

* Publication records support operational troubleshooting, auditability, and @@ -46,16 +49,22 @@ import zeroecho.pki.api.PkiId; *

* * @param publicationId publication id - * @param time time when publication was attempted - * @param target publication target - * @param objectId published object id (credential, CA materials, status - * object) - * @param objectKind non-empty logical kind string (e.g., "CREDENTIAL", - * "CA_MATERIALS", "STATUS_OBJECT") - * @param status publication outcome + * @param sourceType authoritative source-object type + * @param sourceId authoritative source-object identity + * @param content exact immutable source content + * @param target configured destination identity + * @param status durable lifecycle state + * @param attemptNumber non-negative monotonic attempt number + * @param attemptToken stable token for the current attempt, when one exists + * @param createdAt registration time + * @param updatedAt last durable state-change time + * @param failure safe failure classification + * @param evidence safe outcome-evidence classification */ -public record PublicationRecord(PkiId publicationId, Instant time, PublicationTarget target, PkiId objectId, - String objectKind, PublicationStatus status) { +public record PublicationRecord(PkiId publicationId, PublicationSourceType sourceType, PkiId sourceId, + DurableContentReference content, PublicationTarget target, PublicationStatus status, + long attemptNumber, Optional attemptToken, Instant createdAt, Instant updatedAt, + Optional failure, Optional evidence) { /** * Creates a publication record. @@ -63,23 +72,67 @@ public record PublicationRecord(PkiId publicationId, Instant time, PublicationTa * @throws IllegalArgumentException if inputs are invalid */ public PublicationRecord { - if (publicationId == null) { - throw new IllegalArgumentException("publicationId must not be null"); + Objects.requireNonNull(publicationId, "publicationId"); + Objects.requireNonNull(sourceType, "sourceType"); + Objects.requireNonNull(sourceId, "sourceId"); + Objects.requireNonNull(content, "content"); + Objects.requireNonNull(target, "target"); + Objects.requireNonNull(status, "status"); + Objects.requireNonNull(attemptToken, "attemptToken"); + Objects.requireNonNull(createdAt, "createdAt"); + Objects.requireNonNull(updatedAt, "updatedAt"); + Objects.requireNonNull(failure, "failure"); + Objects.requireNonNull(evidence, "evidence"); + if (attemptNumber < 0L || updatedAt.isBefore(createdAt)) { + throw new IllegalArgumentException("Publication revision or time is invalid"); } - if (time == null) { - throw new IllegalArgumentException("time must not be null"); + attemptToken.ifPresent(PublicationRecord::requireAttemptToken); + validateState(status, attemptNumber, attemptToken, failure, evidence); + } + + /** + * Tests whether another record describes the same immutable operation. + * + * @param other candidate record + * @return {@code true} when all immutable fields match exactly + */ + public boolean hasSameOperation(PublicationRecord other) { + return other != null && publicationId.equals(other.publicationId) + && sourceType == other.sourceType && sourceId.equals(other.sourceId) + && sameContent(content, other.content) && target.equals(other.target) + && createdAt.equals(other.createdAt); + } + + private static boolean sameContent(DurableContentReference first, DurableContentReference second) { + return first.storeId().equals(second.storeId()) && first.contentId().equals(second.contentId()) + && first.encoding() == second.encoding() && first.length() == second.length() + && first.sha256().equals(second.sha256()) && first.lifecycle() == second.lifecycle(); + } + + private static void requireAttemptToken(String token) { + if (!token.matches("[0-9a-f]{64}")) { + throw new IllegalArgumentException("Publication attempt token is not canonical"); } - if (target == null) { - throw new IllegalArgumentException("target must not be null"); + } + + private static void validateState(PublicationStatus status, long attemptNumber, Optional token, + Optional failure, Optional evidence) { + boolean hasAttempt = attemptNumber > 0L && token.isPresent(); + switch (status) { + case PENDING -> requireState(token.isEmpty() && failure.isEmpty() && evidence.isEmpty()); + case DISPATCH_PREPARED -> requireState(hasAttempt && failure.isEmpty() && evidence.isEmpty()); + case SUCCEEDED -> requireState(hasAttempt && failure.isEmpty() && evidence.isPresent()); + case RETRYABLE_FAILURE -> requireState(hasAttempt + && failure.filter(value -> value == PublicationFailure.EXTERNAL_RETRYABLE).isPresent() + && evidence.isPresent()); + case TERMINAL_FAILURE -> requireState(hasAttempt && failure.isPresent()); + case OUTCOME_UNKNOWN -> requireState(hasAttempt && failure.isPresent() && evidence.isEmpty()); } - if (objectId == null) { - throw new IllegalArgumentException("objectId must not be null"); - } - if (objectKind == null || objectKind.isBlank()) { - throw new IllegalArgumentException("objectKind must not be null/blank"); - } - if (status == null) { - throw new IllegalArgumentException("status must not be null"); + } + + private static void requireState(boolean valid) { + if (!valid) { + throw new IllegalArgumentException("Publication record state is inconsistent"); } } } diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationRequest.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationRequest.java new file mode 100644 index 0000000..a522e69 --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationRequest.java @@ -0,0 +1,27 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.api.publication; + +import java.util.Objects; +import zeroecho.pki.api.PkiId; + +/** + * Immutable request to register distribution of one already committed PKI object. + * + * @param publicationId caller-selected stable operation identity + * @param sourceType authoritative source-object type + * @param sourceId exact authoritative source-object identity + * @param target configured destination identity + */ +public record PublicationRequest(PkiId publicationId, PublicationSourceType sourceType, + PkiId sourceId, PublicationTarget target) { + /** Creates a validated publication request. */ + public PublicationRequest { + Objects.requireNonNull(publicationId, "publicationId"); + Objects.requireNonNull(sourceType, "sourceType"); + Objects.requireNonNull(sourceId, "sourceId"); + Objects.requireNonNull(target, "target"); + } +} diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationResult.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationResult.java index 070033a..b49634c 100644 --- a/pki/src/main/java/zeroecho/pki/api/publication/PublicationResult.java +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationResult.java @@ -33,8 +33,6 @@ ******************************************************************************/ package zeroecho.pki.api.publication; -import java.util.List; - import zeroecho.pki.api.PkiId; /** @@ -42,9 +40,10 @@ import zeroecho.pki.api.PkiId; * * @param publicationId publication record id * @param status outcome status - * @param notes non-sensitive operator-readable notes + * @param attemptNumber current monotonic attempt number */ -public record PublicationResult(PkiId publicationId, PublicationStatus status, List notes) { +public record PublicationResult(PkiId publicationId, PublicationStatus status, long attemptNumber) { + private static final long INITIAL_ATTEMPT = 0L; /** * Creates a publication result. @@ -58,8 +57,8 @@ public record PublicationResult(PkiId publicationId, PublicationStatus status, L if (status == null) { throw new IllegalArgumentException("status must not be null"); } - if (notes == null) { - throw new IllegalArgumentException("notes must not be null"); + if (attemptNumber < INITIAL_ATTEMPT) { + throw new IllegalArgumentException("attemptNumber must not be negative"); } } } diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationSourceType.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationSourceType.java new file mode 100644 index 0000000..270cae2 --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationSourceType.java @@ -0,0 +1,13 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.api.publication; + +/** Identifies the authoritative PKI object supplying publication content. */ +public enum PublicationSourceType { + /** An immutable committed credential. */ + CREDENTIAL, + /** An immutable committed status object, such as a CRL or OCSP response. */ + STATUS_OBJECT +} diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationStatus.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationStatus.java index e70c3bf..ce88ccd 100644 --- a/pki/src/main/java/zeroecho/pki/api/publication/PublicationStatus.java +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationStatus.java @@ -34,23 +34,31 @@ package zeroecho.pki.api.publication; /** - * Publication outcome status. + * Durable publication-operation state. */ public enum PublicationStatus { /** - * Artifact has been published successfully. + * The operation is registered and eligible for explicit processing. */ - PUBLISHED, + PENDING, /** - * Publication was skipped (e.g., already published, policy decision, target not - * applicable). + * One exact attempt is durably prepared but has no classified outcome. */ - SKIPPED, + DISPATCH_PREPARED, /** - * Publication failed. + * The exact attempted operation has a confirmed successful outcome. */ - FAILED + SUCCEEDED, + + /** The attempt definitively failed and an explicit retry is permitted. */ + RETRYABLE_FAILURE, + + /** The operation definitively failed and is terminal. */ + TERMINAL_FAILURE, + + /** The external effect is unknown and requires explicit reconciliation. */ + OUTCOME_UNKNOWN } diff --git a/pki/src/main/java/zeroecho/pki/api/publication/PublicationTarget.java b/pki/src/main/java/zeroecho/pki/api/publication/PublicationTarget.java index fa27934..022a990 100644 --- a/pki/src/main/java/zeroecho/pki/api/publication/PublicationTarget.java +++ b/pki/src/main/java/zeroecho/pki/api/publication/PublicationTarget.java @@ -33,22 +33,21 @@ ******************************************************************************/ package zeroecho.pki.api.publication; -import zeroecho.pki.api.attr.AttributeSet; - /** * Describes where and how to publish an artifact. * *

- * The {@code targetId} identifies a configured target instance. Additional - * configuration is carried in {@code attributes}. Secrets must not be carried - * in attributes intended for publication. + * The {@code targetId} identifies one administrator-configured publisher. + * Provider configuration and secrets are resolved by that publisher and are + * never carried by this value. *

* * @param type destination type * @param targetId target identifier (implementation-defined) - * @param attributes target configuration/hints (may be empty but not null) */ -public record PublicationTarget(PublicationTargetType type, String targetId, AttributeSet attributes) { +public record PublicationTarget(PublicationTargetType type, String targetId) { + + private static final int MAXIMUM_TARGET_ID_CHARACTERS = 128; /** * Creates a publication target. @@ -59,11 +58,9 @@ public record PublicationTarget(PublicationTargetType type, String targetId, Att if (type == null) { throw new IllegalArgumentException("type must not be null"); } - if (targetId == null || targetId.isBlank()) { - throw new IllegalArgumentException("targetId must not be null/blank"); - } - if (attributes == null) { - throw new IllegalArgumentException("attributes must not be null"); + if (targetId == null || targetId.length() > MAXIMUM_TARGET_ID_CHARACTERS + || !targetId.matches("[A-Za-z0-9][A-Za-z0-9._-]*")) { + throw new IllegalArgumentException("targetId must be a canonical configured identity"); } } } diff --git a/pki/src/main/java/zeroecho/pki/api/publication/package-info.java b/pki/src/main/java/zeroecho/pki/api/publication/package-info.java index 937b1bb..e703d46 100644 --- a/pki/src/main/java/zeroecho/pki/api/publication/package-info.java +++ b/pki/src/main/java/zeroecho/pki/api/publication/package-info.java @@ -35,16 +35,18 @@ * Publication domain model. * *

- * This package defines publication targets and records describing how PKI - * artifacts are distributed to relying parties or infrastructure components - * (repositories, directories, endpoints, etc.). Publication is orchestrated - * through {@link zeroecho.pki.api.PublicationService}. + * This package defines the durable post-commit lifecycle for distributing exact + * immutable credential and status-object content. Publication records are finite + * operational control metadata; they never establish or invalidate PKI authority. + * Processing, retry, and reconciliation are explicit through + * {@link zeroecho.pki.api.PublicationService}. *

* *

Artifacts

*

- * Publication may include certificates, chains, status objects, and related - * metadata. The concrete transport is framework- and deployment-specific. + * Payload bytes remain in the staged-content store and are streamed to one + * configured destination. Unknown external outcomes require reconciliation and + * are never automatically retried. *

* * @since 1.0 diff --git a/pki/src/main/java/zeroecho/pki/impl/core/DefaultPublicationService.java b/pki/src/main/java/zeroecho/pki/impl/core/DefaultPublicationService.java new file mode 100644 index 0000000..8f79947 --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/impl/core/DefaultPublicationService.java @@ -0,0 +1,445 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.impl.core; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Clock; +import java.time.Instant; +import java.util.Collection; +import java.util.HashMap; +import java.util.HexFormat; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import zeroecho.core.io.CancellationSignal; +import zeroecho.core.io.RepeatableContent; +import zeroecho.pki.api.PkiException; +import zeroecho.pki.api.PkiId; +import zeroecho.pki.api.PublicationService; +import zeroecho.pki.api.content.DurableContentReference; +import zeroecho.pki.api.credential.Credential; +import zeroecho.pki.api.publication.PublicationCursor; +import zeroecho.pki.api.publication.PublicationEvidence; +import zeroecho.pki.api.publication.PublicationFailure; +import zeroecho.pki.api.publication.PublicationQuery; +import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationRequest; +import zeroecho.pki.api.publication.PublicationResult; +import zeroecho.pki.api.publication.PublicationSourceType; +import zeroecho.pki.api.publication.PublicationStatus; +import zeroecho.pki.api.publication.PublicationTarget; +import zeroecho.pki.api.status.StatusObject; +import zeroecho.pki.spi.publish.PublicationAttempt; +import zeroecho.pki.spi.publish.PublicationOutcome; +import zeroecho.pki.spi.publish.Publisher; +import zeroecho.pki.spi.store.PkiStore; + +/** + * Durable explicit post-commit publication lifecycle. + * + *

The service is thread-safe. Each external call is fenced by a durable + * compare-and-replace transition and occurs without a metadata transaction or + * store lock. Operations for unrelated identities progress independently. The + * service creates no threads and performs no automatic dispatch or retry.

+ */ +@SuppressWarnings({ "PMD.CyclomaticComplexity", "PMD.AvoidCatchingGenericException", + "PMD.PreserveStackTrace", "PMD.ExceptionAsFlowControl" }) +public final class DefaultPublicationService implements PublicationService { + private final PkiStore store; + private final Clock clock; + private final Map publishers; + private final AtomicBoolean closed = new AtomicBoolean(); + + /** + * Opens a publication lifecycle over one store and immutable publisher registry. + * + *

Opening validates a stable publication snapshot and all configured target + * bindings. The supplied store must already have performed its exclusive-open + * recovery of abandoned dispatch-prepared attempts. Construction never calls + * an external publisher.

+ * + * @param store authoritative PKI store + * @param clock operational state-change clock + * @param publishers configured publisher instances with unique targets + * @throws IllegalArgumentException if a publisher target is duplicated + * @throws PkiException if durable publication recovery fails closed + */ + public DefaultPublicationService(PkiStore store, Clock clock, Collection publishers) { + this.store = Objects.requireNonNull(store, "store"); + this.clock = Objects.requireNonNull(clock, "clock"); + Objects.requireNonNull(publishers, "publishers"); + Map configured = new HashMap<>(); + for (Publisher publisher : publishers) { + Publisher exact = Objects.requireNonNull(publisher, "publisher"); + PublicationTarget target; + try { + target = Objects.requireNonNull(exact.target(), "publisher.target"); + } catch (RuntimeException failure) { + throw new IllegalArgumentException("Publication publisher configuration is invalid"); + } + if (configured.putIfAbsent(target, exact) != null) { + throw new IllegalArgumentException("Publication target is configured more than once"); + } + } + this.publishers = Map.copyOf(configured); + validateConfiguredOperations(); + } + + @Override + public PublicationRecord register(PublicationRequest request) { + requireOpen(); + Objects.requireNonNull(request, "request"); + requirePublisher(request.target()); + DurableContentReference content = resolveSourceContent(request.sourceType(), request.sourceId()); + validateContent(content); + Optional existing = store.getPublicationRecord(request.publicationId()); + if (existing.isPresent()) { + PublicationRecord current = existing.orElseThrow(); + if (current.sourceType() == request.sourceType() && current.sourceId().equals(request.sourceId()) + && current.target().equals(request.target()) && sameContent(current.content(), content)) { + return current; + } + throw new IllegalArgumentException("Publication identity conflicts with an existing operation"); + } + Instant now = clock.instant(); + PublicationRecord pending = new PublicationRecord(request.publicationId(), request.sourceType(), + request.sourceId(), content, request.target(), PublicationStatus.PENDING, 0L, Optional.empty(), + now, now, Optional.empty(), Optional.empty()); + return store.createPublicationRecord(pending); + } + + @Override + public PublicationResult process(PkiId publicationId) { + requireOpen(); + PublicationRecord pending = requireRecord(publicationId); + if (pending.status() != PublicationStatus.PENDING) { + throw new IllegalArgumentException("Publication operation is not pending"); + } + long attemptNumber; + try { + attemptNumber = Math.addExact(pending.attemptNumber(), 1L); + } catch (ArithmeticException overflow) { + throw new PkiException("Publication attempt counter is exhausted: code=PUBLICATION_ATTEMPT_EXHAUSTED"); + } + String token = attemptToken(pending, attemptNumber); + PublicationRecord prepared = copy(pending, PublicationStatus.DISPATCH_PREPARED, attemptNumber, + Optional.of(token), Optional.empty(), Optional.empty()); + if (!store.replacePublicationRecord(pending, prepared)) { + throw new IllegalStateException("Publication operation changed concurrently"); + } + return dispatchPrepared(prepared, PublicationEvidence.DIRECT_RESPONSE, false); + } + + @Override + public PublicationResult retry(PkiId publicationId) { + requireOpen(); + PublicationRecord failed = requireRecord(publicationId); + if (failed.status() != PublicationStatus.RETRYABLE_FAILURE) { + throw new IllegalArgumentException("Publication operation is not retryable"); + } + PublicationRecord pending = copy(failed, PublicationStatus.PENDING, failed.attemptNumber(), + Optional.empty(), Optional.empty(), Optional.empty()); + if (!store.replacePublicationRecord(failed, pending)) { + throw new IllegalStateException("Publication operation changed concurrently"); + } + return process(publicationId); + } + + @Override + public PublicationResult reconcile(PkiId publicationId) { + requireOpen(); + PublicationRecord unknown = requireRecord(publicationId); + if (unknown.status() != PublicationStatus.OUTCOME_UNKNOWN) { + throw new IllegalArgumentException("Publication operation does not require reconciliation"); + } + PublicationRecord prepared = copy(unknown, PublicationStatus.DISPATCH_PREPARED, unknown.attemptNumber(), + unknown.attemptToken(), Optional.empty(), Optional.empty()); + if (!store.replacePublicationRecord(unknown, prepared)) { + throw new IllegalStateException("Publication operation changed concurrently"); + } + return dispatchPrepared(prepared, PublicationEvidence.RECONCILIATION_RESPONSE, true); + } + + @Override + public Optional find(PkiId publicationId) { + requireOpen(); + return store.getPublicationRecord(Objects.requireNonNull(publicationId, "publicationId")); + } + + @Override + public PublicationCursor openPublications(PublicationQuery query) { + requireOpen(); + return new FilteringCursor(store.openPublicationRecords(), Objects.requireNonNull(query, "query")); + } + + @Override + public void close() { + closed.set(true); + } + + private PublicationResult dispatchPrepared(PublicationRecord prepared, PublicationEvidence evidence, + boolean reconciliation) { + Publisher publisher = requirePublisher(prepared.target()); + PublicationAttempt attempt = attempt(prepared); + if (!reconciliation) { + try { + validateContent(prepared.content()); + } catch (PkiException failure) { + PublicationRecord terminal = copy(prepared, PublicationStatus.TERMINAL_FAILURE, + prepared.attemptNumber(), prepared.attemptToken(), + Optional.of(PublicationFailure.CONTENT_INVALID), Optional.empty()); + return persistResult(prepared, terminal); + } + } + + Optional outcome; + if (reconciliation) { + outcome = reconcileExternal(publisher, attempt); + if (outcome.isEmpty()) { + PublicationRecord unresolved = copy(prepared, PublicationStatus.OUTCOME_UNKNOWN, + prepared.attemptNumber(), prepared.attemptToken(), + Optional.of(PublicationFailure.EXTERNAL_OUTCOME_UNKNOWN), Optional.empty()); + return persistResult(prepared, unresolved); + } + } else { + outcome = dispatchExternal(publisher, attempt, prepared); + if (outcome.isEmpty()) { + PublicationRecord unknown = copy(prepared, PublicationStatus.OUTCOME_UNKNOWN, + prepared.attemptNumber(), prepared.attemptToken(), + Optional.of(PublicationFailure.EXTERNAL_OUTCOME_UNKNOWN), Optional.empty()); + return persistResult(prepared, unknown); + } + } + PublicationRecord classified = classified(prepared, outcome.orElseThrow(), evidence); + return persistResult(prepared, classified); + } + + private Optional dispatchExternal(Publisher publisher, PublicationAttempt attempt, + PublicationRecord prepared) { + try (RepeatableContent payload = store.stagedContent().openContent(prepared.content())) { + return Optional.of(Objects.requireNonNull(publisher.publish(attempt, payload), "publisher outcome")); + } catch (IOException | RuntimeException failure) { + return Optional.empty(); + } + } + + private static Optional reconcileExternal(Publisher publisher, PublicationAttempt attempt) { + try { + return Objects.requireNonNull(publisher.reconcile(attempt), "publisher reconciliation outcome"); + } catch (RuntimeException failure) { + return Optional.empty(); + } + } + + private PublicationResult persistResult(PublicationRecord prepared, PublicationRecord classified) { + try { + if (store.replacePublicationRecord(prepared, classified)) { + return result(classified); + } + PublicationRecord observed = requireRecord(prepared.publicationId()); + if (observed.status() != PublicationStatus.DISPATCH_PREPARED) { + return result(observed); + } + } catch (RuntimeException persistenceFailure) { + Optional observed = safeFind(prepared.publicationId()); + if (observed.isPresent() && observed.orElseThrow().status() != PublicationStatus.DISPATCH_PREPARED) { + return result(observed.orElseThrow()); + } + } + PublicationRecord unknown = copy(prepared, PublicationStatus.OUTCOME_UNKNOWN, prepared.attemptNumber(), + prepared.attemptToken(), Optional.of(PublicationFailure.RESULT_PERSISTENCE_UNKNOWN), Optional.empty()); + try { + store.replacePublicationRecord(prepared, unknown); + } catch (RuntimeException ignored) { + // The durable PREPARED fence remains non-dispatchable and is converted + // to OUTCOME_UNKNOWN during the next service recovery. + } + throw new PkiException( + "Publication result persistence requires reconciliation: code=PUBLICATION_RESULT_UNCONFIRMED"); + } + + private Optional safeFind(PkiId publicationId) { + try { + return store.getPublicationRecord(publicationId); + } catch (RuntimeException ignored) { + return Optional.empty(); + } + } + + private void validateConfiguredOperations() { + try (PublicationCursor cursor = store.openPublicationRecords()) { + Optional next; + while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) { + PublicationRecord record = next.orElseThrow(); + requirePublisher(record.target()); + } + } catch (IOException | RuntimeException failure) { + throw publicationRecoveryFailure(); + } + } + + private DurableContentReference resolveSourceContent(PublicationSourceType type, PkiId sourceId) { + Objects.requireNonNull(type, "sourceType"); + Objects.requireNonNull(sourceId, "sourceId"); + return switch (type) { + case CREDENTIAL -> store.getCredential(sourceId).map(Credential::content) + .orElseThrow(() -> new IllegalArgumentException("Publication credential is not committed")); + case STATUS_OBJECT -> store.getStatusObject(sourceId).map(StatusObject::content) + .orElseThrow(() -> new IllegalArgumentException("Publication status object is not committed")); + }; + } + + private void validateContent(DurableContentReference reference) { + try { + DurableContentReference exact = store.stagedContent().restoreReference(reference.storeId(), + reference.contentId(), reference.encoding(), reference.length(), reference.sha256(), + reference.lifecycle()); + if (!sameContent(reference, exact)) { + throw new IOException("Publication content commitment mismatch"); + } + try (RepeatableContent content = store.stagedContent().openContent(exact); + java.io.InputStream ignored = content.openStream()) { + // Opening the store-issued stream performs the complete immutable + // integrity check before any publisher callback can run. + } + } catch (IOException | RuntimeException failure) { + throw new PkiException("Publication content is invalid: code=PUBLICATION_CONTENT_INVALID"); + } + } + + private Publisher requirePublisher(PublicationTarget target) { + Publisher publisher = publishers.get(Objects.requireNonNull(target, "target")); + if (publisher == null) { + throw new IllegalArgumentException("Publication target is not configured"); + } + return publisher; + } + + private PublicationRecord requireRecord(PkiId publicationId) { + return store.getPublicationRecord(Objects.requireNonNull(publicationId, "publicationId")) + .orElseThrow(() -> new IllegalArgumentException("Publication operation does not exist")); + } + + private PublicationRecord copy(PublicationRecord current, PublicationStatus status, long attemptNumber, + Optional token, Optional failure, Optional evidence) { + Instant updatedAt = clock.instant(); + if (updatedAt.isBefore(current.updatedAt())) { + updatedAt = current.updatedAt(); + } + return new PublicationRecord(current.publicationId(), current.sourceType(), current.sourceId(), + current.content(), current.target(), status, attemptNumber, token, current.createdAt(), updatedAt, + failure, evidence); + } + + private PublicationRecord classified(PublicationRecord prepared, PublicationOutcome outcome, + PublicationEvidence evidence) { + return switch (outcome) { + case SUCCESS -> copy(prepared, PublicationStatus.SUCCEEDED, prepared.attemptNumber(), + prepared.attemptToken(), Optional.empty(), Optional.of(evidence)); + case RETRYABLE_FAILURE -> copy(prepared, PublicationStatus.RETRYABLE_FAILURE, prepared.attemptNumber(), + prepared.attemptToken(), Optional.of(PublicationFailure.EXTERNAL_RETRYABLE), + Optional.of(evidence)); + case TERMINAL_FAILURE -> copy(prepared, PublicationStatus.TERMINAL_FAILURE, prepared.attemptNumber(), + prepared.attemptToken(), Optional.of(PublicationFailure.EXTERNAL_TERMINAL), + Optional.of(evidence)); + }; + } + + private static PublicationAttempt attempt(PublicationRecord prepared) { + DurableContentReference content = prepared.content(); + return new PublicationAttempt(prepared.publicationId(), prepared.sourceType(), prepared.sourceId(), + prepared.target(), prepared.attemptNumber(), prepared.attemptToken().orElseThrow(), + content.encoding(), content.length(), content.sha256()); + } + + private static PublicationResult result(PublicationRecord record) { + return new PublicationResult(record.publicationId(), record.status(), record.attemptNumber()); + } + + private static String attemptToken(PublicationRecord record, long attemptNumber) { + MessageDigest digest = sha256(); + update(digest, record.publicationId().value()); + update(digest, record.sourceType().name()); + update(digest, record.sourceId().value()); + update(digest, record.target().type().name()); + update(digest, record.target().targetId()); + update(digest, record.content().encoding().name()); + update(digest, record.content().sha256()); + digest.update(ByteBuffer.allocate(Long.BYTES).putLong(record.content().length()).array()); + digest.update(ByteBuffer.allocate(Long.BYTES).putLong(attemptNumber).array()); + return HexFormat.of().formatHex(digest.digest()); + } + + private static void update(MessageDigest digest, String value) { + byte[] encoded = value.getBytes(StandardCharsets.UTF_8); + digest.update(ByteBuffer.allocate(Integer.BYTES).putInt(encoded.length).array()); + digest.update(encoded); + } + + private static MessageDigest sha256() { + try { + return MessageDigest.getInstance("SHA-256"); + } catch (NoSuchAlgorithmException impossible) { + throw new IllegalStateException("SHA-256 is unavailable", impossible); + } + } + + private static boolean sameContent(DurableContentReference first, DurableContentReference second) { + return first.storeId().equals(second.storeId()) && first.contentId().equals(second.contentId()) + && first.encoding() == second.encoding() && first.length() == second.length() + && first.sha256().equals(second.sha256()) && first.lifecycle() == second.lifecycle(); + } + + private void requireOpen() { + if (closed.get()) { + throw new IllegalStateException("Publication service is closed"); + } + } + + private static PkiException publicationRecoveryFailure() { + return new PkiException("Publication recovery failed: code=PUBLICATION_RECOVERY_FAILED"); + } + + /** Lazy query wrapper retaining the stable store cursor. */ + private static final class FilteringCursor implements PublicationCursor { + private final PublicationCursor delegate; + private final PublicationQuery query; + + private FilteringCursor(PublicationCursor delegate, PublicationQuery query) { + this.delegate = delegate; + this.query = query; + } + + @Override + public Optional next(CancellationSignal cancellation) throws IOException { + Optional next; + while ((next = delegate.next(cancellation)).isPresent()) { + PublicationRecord record = next.orElseThrow(); + boolean targetMatch = query.targetType().isEmpty() + || query.targetType().orElseThrow() == record.target().type(); + boolean afterMatch = query.after().isEmpty() + || !record.createdAt().isBefore(query.after().orElseThrow()); + boolean beforeMatch = query.before().isEmpty() + || !record.createdAt().isAfter(query.before().orElseThrow()); + boolean sourceMatch = query.sourceType().isEmpty() + || query.sourceType().orElseThrow() == record.sourceType(); + if (targetMatch && afterMatch && beforeMatch && sourceMatch) { + return next; + } + } + return Optional.empty(); + } + + @Override + public void close() { + delegate.close(); + } + } +} diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java b/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java index 1e15080..169e0bc 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java @@ -84,7 +84,11 @@ import zeroecho.pki.api.policy.PolicyTrace; import zeroecho.pki.api.profile.ActiveCertificateProfile; import zeroecho.pki.api.profile.CertificateProfileRef; import zeroecho.pki.api.profile.ImportedCertificateProfileVersion; +import zeroecho.pki.api.publication.PublicationCursor; +import zeroecho.pki.api.publication.PublicationFailure; import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationSourceType; +import zeroecho.pki.api.publication.PublicationStatus; import zeroecho.pki.api.request.ParsedCertificationRequest; import zeroecho.pki.api.revocation.RevocationCommand; import zeroecho.pki.api.revocation.RevocationRecord; @@ -173,11 +177,13 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { private static final String SIGN_OWNER_NAMESPACE = "io.zeroecho.pki.signing-owner"; private static final String STATUS_RECORD_NAMESPACE = "io.zeroecho.pki.status-object-record"; private static final String STATUS_OWNER_NAMESPACE = "io.zeroecho.pki.status-object-owner"; + private static final String PUBLICATION_RECORD_NAMESPACE = "io.zeroecho.pki.publication-record"; private static final int CURRENT_SIGN_RECORD_VERSION = 2; private static final int SIGN_OWNER_VALUE_VERSION = 1; private static final int STATUS_OWNER_VALUE_VERSION = 1; private static final int METADATA_TRANSFER_BUFFER_BYTES = 16 * 1024; private static final ThreadLocal STATUS_COMMIT_FAULT = new ThreadLocal<>(); + private static final ThreadLocal PUBLICATION_COMMIT_FAULT = new ThreadLocal<>(); private static final String SIGN_FINGERPRINT_PREFIX = "signfp:v1:"; private static final int SIGN_FINGERPRINT_HEX_LENGTH = 64; private static final long INITIAL_FENCE = 0L; @@ -268,6 +274,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { rejectNonEmptyUnversionedStore(); } ensureVersionFile(); + rejectObsoletePublicationLayout(); this.signingNamespace = ensureSigningNamespace(); this.stagedContent = new FilesystemStagedContentStore(this.paths.stagedContentRoot(), this.signingNamespace); @@ -279,6 +286,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { this.signingTimeWatermark = new AtomicLong(loadSigningTimeWatermark()); this.historySeq = new AtomicLong(0L); recoverStagedContent(); + recoverPublicationRecords(); boolean snapshotRestore = requireSnapshotBoundary(); openedRevocations = FilesystemRevocationAuthority.open( this.paths, new MetadataStoreId(this.signingNamespace), credentialId -> { @@ -757,19 +765,117 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } @Override - public void putPublicationRecord(final PublicationRecord record) { + public PublicationRecord createPublicationRecord(final PublicationRecord record) { requireStoreUsable(); Objects.requireNonNull(record, "record"); - PkiId id = record.publicationId(); - writeOnce(this.paths.publicationPath(id), FsCodec.encode(FsCodec.PUBLICATION, record), "PUBLICATION", - FsUtil.safeId(id)); + validatePublicationRecord(record); + Optional existing = getPublicationRecord(record.publicationId()); + if (existing.isPresent()) { + if (publicationRecordsEqual(existing.orElseThrow(), record)) { + return existing.orElseThrow(); + } + throw new IllegalArgumentException("Publication identity conflicts with an existing operation"); + } + try (MetadataTransaction transaction = metadataStore.beginTransaction()) { + transaction.create(publicationRecordKey(record.publicationId()), + byteContent(PublicationRecordCodec.encode(record)), CancellationSignal.NONE); + PublicationCommitFaultPoint fault = takePublicationCommitFault(); + if (fault == PublicationCommitFaultPoint.CREATE_BEFORE_COMMIT) { + throw new IOException("Injected publication create failure"); + } + MetadataCommitResult result = transaction.commit(); + if (result.outcome() == MetadataCommitResult.Outcome.COMMITTED) { + return record; + } + if (result.outcome() == MetadataCommitResult.Outcome.UNKNOWN) { + throw publicationPersistenceUnknown(); + } + } catch (IOException exception) { + throw new IllegalStateException("Publication metadata persistence failed"); + } + Optional raced = getPublicationRecord(record.publicationId()); + if (raced.isPresent() && publicationRecordsEqual(raced.orElseThrow(), record)) { + return raced.orElseThrow(); + } + throw new IllegalArgumentException("Publication identity conflicts with an existing operation"); } @Override - public List listPublicationRecords() { + public Optional getPublicationRecord(final PkiId publicationId) { requireStoreUsable(); - Path byId = this.paths.root().resolve("publications").resolve("by-id"); - return listBinaryFiles(byId, FsCodec.PUBLICATION); + Objects.requireNonNull(publicationId, "publicationId"); + MetadataKey key = publicationRecordKey(publicationId); + try (MetadataSnapshot snapshot = metadataStore.snapshot()) { + Optional stored = snapshot.get(key); + if (stored.isEmpty()) { + return Optional.empty(); + } + try (MetadataSnapshot.Record record = stored.orElseThrow()) { + return Optional.of(decodePublicationRecord(record, snapshot)); + } + } catch (IOException exception) { + throw new IllegalStateException("Publication metadata is invalid"); + } + } + + @Override + public boolean replacePublicationRecord(final PublicationRecord expected, + final PublicationRecord replacement) { + requireStoreUsable(); + Objects.requireNonNull(expected, "expected"); + Objects.requireNonNull(replacement, "replacement"); + validatePublicationReplacement(expected, replacement); + MetadataKey key = publicationRecordKey(expected.publicationId()); + try (MetadataSnapshot snapshot = metadataStore.snapshot()) { + Optional stored = snapshot.get(key); + if (stored.isEmpty()) { + return false; + } + long revision; + try (MetadataSnapshot.Record record = stored.orElseThrow()) { + if (!publicationRecordsEqual(decodePublicationRecord(record, snapshot), expected)) { + return false; + } + revision = record.recordRevision(); + } + try (MetadataTransaction transaction = metadataStore.beginTransaction()) { + transaction.replace(key, revision, byteContent(PublicationRecordCodec.encode(replacement)), + CancellationSignal.NONE); + PublicationCommitFaultPoint fault = takePublicationCommitFault(); + if (fault == PublicationCommitFaultPoint.REPLACE_BEFORE_COMMIT) { + throw new IOException("Injected publication replacement failure"); + } + MetadataCommitResult result = transaction.commit(); + if (result.outcome() == MetadataCommitResult.Outcome.COMMITTED + && fault == PublicationCommitFaultPoint.REPLACE_AFTER_COMMIT_AS_UNKNOWN) { + throw publicationPersistenceUnknown(); + } + if (result.outcome() == MetadataCommitResult.Outcome.UNKNOWN) { + throw publicationPersistenceUnknown(); + } + return result.outcome() == MetadataCommitResult.Outcome.COMMITTED; + } + } catch (IOException exception) { + throw new IllegalStateException("Publication metadata persistence failed"); + } + } + + @Override + @SuppressWarnings({ "PMD.AvoidCatchingGenericException", "PMD.PreserveStackTrace" }) + public PublicationCursor openPublicationRecords() { + requireStoreUsable(); + MetadataSnapshot snapshot = null; + try { + snapshot = metadataStore.snapshot(); + MetadataCursor cursor = snapshot.scan(MetadataSnapshot.KeyRange.all(PUBLICATION_RECORD_NAMESPACE), + CancellationSignal.NONE); + return new PublicationMetadataCursor(snapshot, cursor); + } catch (IOException | RuntimeException failure) { + if (snapshot != null) { + snapshot.close(); + } + throw new IllegalStateException("Publication metadata cursor cannot be opened"); + } } @Override @@ -1827,6 +1933,162 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { return new MetadataKey(STATUS_OWNER_NAMESPACE, statusObjectId.value()); } + private static MetadataKey publicationRecordKey(PkiId publicationId) { + return new MetadataKey(PUBLICATION_RECORD_NAMESPACE, publicationId.value()); + } + + private PublicationRecord decodePublicationRecord(MetadataSnapshot.Record stored, MetadataSnapshot snapshot) + throws IOException { + PublicationRecord record = PublicationRecordCodec.decode(readMetadataValue(stored), stagedContent); + if (!publicationRecordKey(record.publicationId()).equals(stored.key())) { + throw new IOException("Publication metadata key mismatch"); + } + validatePublicationRecord(record, snapshot); + return record; + } + + @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops") + private void recoverPublicationRecords() { + try (PublicationCursor cursor = openPublicationRecords()) { + Optional next; + while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) { + PublicationRecord record = next.orElseThrow(); + if (record.status() == PublicationStatus.DISPATCH_PREPARED) { + Instant observed = clock.instant(); + Instant updatedAt = observed.isAfter(record.updatedAt()) ? observed : record.updatedAt(); + PublicationRecord unknown = new PublicationRecord(record.publicationId(), record.sourceType(), + record.sourceId(), record.content(), record.target(), PublicationStatus.OUTCOME_UNKNOWN, + record.attemptNumber(), record.attemptToken(), record.createdAt(), updatedAt, + Optional.of(PublicationFailure.EXTERNAL_OUTCOME_UNKNOWN), Optional.empty()); + if (!replacePublicationRecord(record, unknown)) { + throw new IllegalStateException("Publication recovery conflict"); + } + } + } + } catch (IOException exception) { + throw new IllegalStateException("Publication recovery failed"); + } + } + + private void validatePublicationRecord(PublicationRecord record) { + try (MetadataSnapshot snapshot = metadataStore.snapshot()) { + validatePublicationRecord(record, snapshot); + } catch (IOException exception) { + throw new IllegalStateException("Publication source authority is invalid"); + } + } + + private void validatePublicationRecord(PublicationRecord record, MetadataSnapshot snapshot) throws IOException { + publicationRecordKey(record.publicationId()); + DurableContentReference reference = record.content(); + if (!stagedContent.contentStoreId().equals(reference.storeId()) + || reference.lifecycle() != DurableContentReference.Lifecycle.PERSISTED) { + throw new IllegalArgumentException("Publication content is not persistent store authority"); + } + try { + stagedContent.restoreReference(reference.storeId(), reference.contentId(), reference.encoding(), + reference.length(), reference.sha256(), reference.lifecycle()); + } catch (IOException exception) { + throw new IllegalStateException("Publication content commitment is invalid"); + } + DurableContentReference authoritative = record.sourceType() == PublicationSourceType.CREDENTIAL + ? getCredential(record.sourceId()).map(Credential::content).orElseThrow( + () -> new IllegalStateException("Publication credential source is missing")) + : statusContent(snapshot, record.sourceId()); + if (!sameContentReference(authoritative, reference)) { + throw new IllegalArgumentException("Publication content does not match its source object"); + } + } + + private DurableContentReference statusContent(MetadataSnapshot snapshot, PkiId statusId) throws IOException { + Optional stored = snapshot.get(statusRecordKey(statusId)); + if (stored.isPresent()) { + return decodeStoredStatus(snapshot, stored.orElseThrow()).status().content(); + } + Optional owner = snapshot.get(statusOwnerKey(statusId)); + if (owner.isPresent()) { + try (MetadataSnapshot.Record ignored = owner.orElseThrow()) { + throw new IOException("Status owner exists without its record"); + } + } + throw new IllegalStateException("Publication status source is missing"); + } + + private static void validatePublicationReplacement(PublicationRecord expected, PublicationRecord replacement) { + if (!expected.hasSameOperation(replacement) + || replacement.updatedAt().isBefore(expected.updatedAt())) { + throw new IllegalArgumentException("Publication immutable fields cannot change"); + } + PublicationStatus previous = expected.status(); + PublicationStatus next = replacement.status(); + boolean valid = switch (previous) { + case PENDING -> next == PublicationStatus.DISPATCH_PREPARED + && replacement.attemptNumber() == Math.addExact(expected.attemptNumber(), 1L); + case DISPATCH_PREPARED -> (next == PublicationStatus.SUCCEEDED + || next == PublicationStatus.RETRYABLE_FAILURE + || next == PublicationStatus.TERMINAL_FAILURE + || next == PublicationStatus.OUTCOME_UNKNOWN) + && sameAttempt(expected, replacement); + case RETRYABLE_FAILURE -> next == PublicationStatus.PENDING + && replacement.attemptNumber() == expected.attemptNumber(); + case OUTCOME_UNKNOWN -> next == PublicationStatus.DISPATCH_PREPARED + && sameAttempt(expected, replacement); + case SUCCEEDED, TERMINAL_FAILURE -> false; + }; + if (!valid) { + throw new IllegalArgumentException("Publication state transition is illegal"); + } + } + + private static boolean sameAttempt(PublicationRecord first, PublicationRecord second) { + return first.attemptNumber() == second.attemptNumber() + && first.attemptToken().equals(second.attemptToken()); + } + + private static boolean publicationRecordsEqual(PublicationRecord first, PublicationRecord second) { + return first.hasSameOperation(second) && second.hasSameOperation(first) + && first.status() == second.status() && first.attemptNumber() == second.attemptNumber() + && first.attemptToken().equals(second.attemptToken()) + && first.updatedAt().equals(second.updatedAt()) && first.failure().equals(second.failure()) + && first.evidence().equals(second.evidence()); + } + + private static boolean sameContentReference(DurableContentReference first, DurableContentReference second) { + return first.storeId().equals(second.storeId()) && first.contentId().equals(second.contentId()) + && first.encoding() == second.encoding() && first.length() == second.length() + && first.sha256().equals(second.sha256()) && first.lifecycle() == second.lifecycle(); + } + + private static PkiException publicationPersistenceUnknown() { + return new PkiException( + "Publication metadata outcome requires recovery: code=PUBLICATION_PERSISTENCE_UNKNOWN"); + } + + /* package */ static void installPublicationCommitFault(PublicationCommitFaultPoint point) { + PUBLICATION_COMMIT_FAULT.set(Objects.requireNonNull(point, "point")); + } + + /* package */ static void clearPublicationCommitFault() { + PUBLICATION_COMMIT_FAULT.remove(); + } + + private static PublicationCommitFaultPoint takePublicationCommitFault() { + PublicationCommitFaultPoint point = PUBLICATION_COMMIT_FAULT.get(); + PUBLICATION_COMMIT_FAULT.remove(); + return point; + } + + /** Test-only publication metadata commit boundary. */ + /* default */ + enum PublicationCommitFaultPoint { + /** Fails registration before the metadata commit attempt. */ + CREATE_BEFORE_COMMIT, + /** Fails one state replacement before the metadata commit attempt. */ + REPLACE_BEFORE_COMMIT, + /** Reports uncertainty after one replacement was durably committed. */ + REPLACE_AFTER_COMMIT_AS_UNKNOWN + } + private static RepeatableContent byteContent(byte[] value) { return new ByteValueContent(value); } @@ -2349,6 +2611,12 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } } + private void rejectObsoletePublicationLayout() { + if (Files.exists(this.paths.root().resolve("publications"))) { + throw new IllegalStateException("obsolete publication layout is not supported"); + } + } + private Optional readOptional(final Path path, final FsCodec.Schema schema) { try { if (!Files.exists(path)) { @@ -2361,24 +2629,6 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } } - private List listBinaryFiles(final Path byIdDir, final FsCodec.Schema schema) { - if (!Files.isDirectory(byIdDir)) { - return List.of(); - } - try { - return Files.list(byIdDir).filter(Files::isRegularFile) - .sorted(Comparator.comparing(p -> p.getFileName().toString())).map(p -> { - try { - return FsCodec.decode(schema, FsOperations.readAll(p), stagedContent); - } catch (IOException e) { - throw new IllegalStateException("read failed: " + p, e); - } - }).toList(); - } catch (IOException e) { - throw new IllegalStateException("list failed: " + byIdDir, e); - } - } - private List listCurrentRecords(final Path byIdDir, final FsCodec.Schema schema) { if (!Files.isDirectory(byIdDir)) { return List.of(); @@ -2479,4 +2729,39 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { // deterministic best-effort } } + + /** Stable metadata cursor that decodes and validates one record at a time. */ + private final class PublicationMetadataCursor implements PublicationCursor { + private final MetadataSnapshot snapshot; + private final MetadataCursor cursor; + private final AtomicBoolean cursorClosed = new AtomicBoolean(); + + private PublicationMetadataCursor(MetadataSnapshot snapshot, MetadataCursor cursor) { + this.snapshot = snapshot; + this.cursor = cursor; + } + + @Override + public Optional next(CancellationSignal cancellation) throws IOException { + Objects.requireNonNull(cancellation, "cancellation"); + if (cursorClosed.get()) { + throw new IllegalStateException("Publication cursor is closed"); + } + Optional next = cursor.next(cancellation); + if (next.isEmpty()) { + return Optional.empty(); + } + try (MetadataSnapshot.Record stored = next.orElseThrow()) { + return Optional.of(decodePublicationRecord(stored, snapshot)); + } + } + + @Override + public void close() { + if (cursorClosed.compareAndSet(false, true)) { + cursor.close(); + snapshot.close(); + } + } + } } diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FsCodec.java b/pki/src/main/java/zeroecho/pki/impl/fs/FsCodec.java index 5a6187c..13d8b33 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FsCodec.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FsCodec.java @@ -80,10 +80,6 @@ import zeroecho.pki.api.profile.CertificateProfileRef; import zeroecho.pki.api.profile.ImportedCertificateProfileVersion; import zeroecho.pki.api.profile.SubjectAlternativeNameType; import zeroecho.pki.api.profile.SubjectRdnType; -import zeroecho.pki.api.publication.PublicationRecord; -import zeroecho.pki.api.publication.PublicationStatus; -import zeroecho.pki.api.publication.PublicationTarget; -import zeroecho.pki.api.publication.PublicationTargetType; import zeroecho.pki.api.request.ParsedCertificationRequest; import zeroecho.pki.api.request.SubjectAlternativeName; import zeroecho.pki.api.request.SubjectRdn; @@ -125,7 +121,6 @@ final class FsCodec { private static final int TOP_CREDENTIAL = 2; private static final int TOP_PARSED_REQUEST = 3; private static final int TOP_STATUS_OBJECT = 5; - private static final int TOP_PUBLICATION = 6; private static final int TOP_POLICY_TRACE = 8; private static final int TOP_WORKFLOW_STATE = 9; private static final int TOP_SIGN_WORKFLOW_RECORD = 10; @@ -150,7 +145,6 @@ final class FsCodec { private static final int TYPE_PRINCIPAL = 27; private static final int TYPE_ATTRIBUTE_ID = 28; private static final int TYPE_POLICY_TRACE_STEP = 29; - private static final int TYPE_PUBLICATION_TARGET = 30; private static final int TYPE_ATTRIBUTE_SET = 31; private static final int TYPE_ATTRIBUTE_VALUE = 33; private static final int TYPE_CREDENTIAL_RECORD = 34; @@ -159,8 +153,6 @@ final class FsCodec { private static final int TYPE_CA_STATE_ENUM = 52; private static final int TYPE_CREDENTIAL_STATUS_ENUM = 53; private static final int TYPE_STATUS_OBJECT_TYPE_ENUM = 55; - private static final int TYPE_PUBLICATION_TARGET_TYPE_ENUM = 56; - private static final int TYPE_PUBLICATION_STATUS_ENUM = 57; private static final int TYPE_DURABILITY_POLICY_ENUM = 58; private static final int TYPE_SIGN_STATE_ENUM = 59; private static final int TYPE_SUBJECT_RDN_TYPE_ENUM = 62; @@ -250,32 +242,6 @@ final class FsCodec { case 4 -> StatusObjectType.REVOCATION_LIST; default -> throw unknownEnum("StatusObjectType", code); }); - private static final ValueSchema PUBLICATION_TARGET_TYPE = enumSchema( - TYPE_PUBLICATION_TARGET_TYPE_ENUM, value -> switch (value) { - case FILESYSTEM -> 1; - case LDAP -> 2; - case HTTP -> 3; - case OBJECT_STORE -> 4; - case CUSTOM -> 5; - }, code -> switch (code) { - case 1 -> PublicationTargetType.FILESYSTEM; - case 2 -> PublicationTargetType.LDAP; - case 3 -> PublicationTargetType.HTTP; - case 4 -> PublicationTargetType.OBJECT_STORE; - case 5 -> PublicationTargetType.CUSTOM; - default -> throw unknownEnum("PublicationTargetType", code); - }); - private static final ValueSchema PUBLICATION_STATUS = enumSchema(TYPE_PUBLICATION_STATUS_ENUM, - value -> switch (value) { - case PUBLISHED -> 1; - case SKIPPED -> 2; - case FAILED -> 3; - }, code -> switch (code) { - case 1 -> PublicationStatus.PUBLISHED; - case 2 -> PublicationStatus.SKIPPED; - case 3 -> PublicationStatus.FAILED; - default -> throw unknownEnum("PublicationStatus", code); - }); private static final ValueSchema DURABILITY_POLICY = enumSchema( TYPE_DURABILITY_POLICY_ENUM, value -> switch (value) { case STRICT_ABORT_ON_RESTART -> 1; @@ -384,14 +350,6 @@ final class FsCodec { }, reader -> new PolicyTraceStep(reader.readValue(STRING), reader.readValue(STRING), reader.readValue(STRINGS))); private static final ValueSchema> POLICY_TRACE_STEPS = listOf(POLICY_TRACE_STEP); - private static final ValueSchema PUBLICATION_TARGET = valueSchema(TYPE_PUBLICATION_TARGET, - (writer, value) -> { - writer.writeValue(PUBLICATION_TARGET_TYPE, value.type()); - writer.writeValue(STRING, value.targetId()); - writer.writeValue(ATTRIBUTE_SET, value.attributes()); - }, reader -> new PublicationTarget(reader.readValue(PUBLICATION_TARGET_TYPE), reader.readValue(STRING), - reader.readValue(ATTRIBUTE_SET))); - private static final ValueSchema> OPTIONAL_VALIDITY = optionalOf(VALIDITY); private static final ValueSchema> OPTIONAL_STRING = optionalOf(STRING); private static final ValueSchema> OPTIONAL_INSTANT = optionalOf(INSTANT); @@ -408,8 +366,6 @@ final class FsCodec { "PARSED_REQUEST", valueSchema(101, FsCodec::writeParsedRequest, FsCodec::readParsedRequest)); /* package */ static final Schema STATUS_OBJECT = topLevel(TOP_STATUS_OBJECT, "STATUS_OBJECT", valueSchema(103, FsCodec::writeStatusObject, FsCodec::readStatusObject)); - /* package */ static final Schema PUBLICATION = topLevel(TOP_PUBLICATION, "PUBLICATION", - valueSchema(104, FsCodec::writePublication, FsCodec::readPublication)); /* package */ static final Schema POLICY_TRACE = topLevel(TOP_POLICY_TRACE, "POLICY_TRACE", valueSchema(106, FsCodec::writePolicyTrace, FsCodec::readPolicyTrace)); /* package */ static final Schema WORKFLOW_STATE = topLevel(TOP_WORKFLOW_STATE, @@ -425,7 +381,7 @@ final class FsCodec { private static final Map> TOP_LEVEL_SCHEMAS = Map.ofEntries(Map.entry(TOP_CA_RECORD, CA_RECORD), Map.entry(TOP_CREDENTIAL, CREDENTIAL), Map.entry(TOP_PARSED_REQUEST, PARSED_REQUEST), Map.entry(TOP_STATUS_OBJECT, STATUS_OBJECT), - Map.entry(TOP_PUBLICATION, PUBLICATION), Map.entry(TOP_POLICY_TRACE, POLICY_TRACE), + Map.entry(TOP_POLICY_TRACE, POLICY_TRACE), Map.entry(TOP_WORKFLOW_STATE, WORKFLOW_STATE), Map.entry(TOP_SIGN_WORKFLOW_RECORD, SIGN_WORKFLOW_RECORD), Map.entry(TOP_PROFILE_VERSION, PROFILE_VERSION), Map.entry(TOP_ACTIVE_PROFILE_REF, ACTIVE_PROFILE_REF)); @@ -711,21 +667,6 @@ final class FsCodec { reader.readValue(ATTRIBUTE_SET)); } - private static void writePublication(Writer writer, PublicationRecord value) throws IOException { - writer.writeValue(PKI_ID, value.publicationId()); - writer.writeValue(INSTANT, value.time()); - writer.writeValue(PUBLICATION_TARGET, value.target()); - writer.writeValue(PKI_ID, value.objectId()); - writer.writeValue(STRING, value.objectKind()); - writer.writeValue(PUBLICATION_STATUS, value.status()); - } - - private static PublicationRecord readPublication(Reader reader) throws IOException { - return new PublicationRecord(reader.readValue(PKI_ID), reader.readValue(INSTANT), - reader.readValue(PUBLICATION_TARGET), reader.readValue(PKI_ID), reader.readValue(STRING), - reader.readValue(PUBLICATION_STATUS)); - } - private static void writeProfileVersion(Writer writer, ImportedCertificateProfileVersion value) throws IOException { writer.writeValue(PROFILE_REF, value.reference()); writer.writeValue(LONG, (long) value.schemaVersion()); diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java b/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java index 8e8e6f2..44a1b28 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java @@ -199,15 +199,6 @@ final class FsPaths { return this.root.resolve("policy").resolve("by-decision").resolve(FsUtil.safeId(decisionId) + ".bin"); } - // ------------------------------------------------------------------------- - // Publications (immutable .bin) - // ------------------------------------------------------------------------- - - /* default */ Path publicationPath(final PkiId publicationId) { - Objects.requireNonNull(publicationId, "publicationId"); - return this.root.resolve("publications").resolve(BY_ID).resolve(FsUtil.safeId(publicationId) + ".bin"); - } - // ------------------------------------------------------------------------- // Workflow state (mutable with history) // ------------------------------------------------------------------------- diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FsSnapshotExporter.java b/pki/src/main/java/zeroecho/pki/impl/fs/FsSnapshotExporter.java index 2bb3e53..f3331b1 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FsSnapshotExporter.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FsSnapshotExporter.java @@ -51,15 +51,20 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.logging.Level; import java.util.logging.Logger; +import zeroecho.core.io.CancellationSignal; import zeroecho.core.io.RepeatableContent; import zeroecho.pki.api.PkiId; import zeroecho.pki.api.ca.CaRecord; import zeroecho.pki.api.content.DurableContentReference; import zeroecho.pki.api.credential.Credential; +import zeroecho.pki.api.publication.PublicationCursor; +import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationSourceType; import zeroecho.pki.api.status.StatusObject; import zeroecho.pki.impl.ProfileLifecycleFailure; import zeroecho.pki.impl.ProfileLifecycleFailure.Code; @@ -337,27 +342,48 @@ final class FsSnapshotExporter { } private void build(Path targetRoot, Instant at) throws IOException { - Path sourceRoot = source.snapshotRoot(); - Path stagedRevocationPrefix = targetRoot.resolve(REVOCATION_PREFIX_STAGE); - FilesystemRevocationLog.RecoveryTarget captured = - source.copyRevocationPrefix(stagedRevocationPrefix); - copyFile(sourceRoot.resolve("SIGNING_NAMESPACE"), targetRoot.resolve("SIGNING_NAMESPACE")); - copyFile(sourceRoot.resolve("SIGNING_TIME_WATERMARK"), targetRoot.resolve("SIGNING_TIME_WATERMARK")); - copyTreeIfExists(sourceRoot.resolve("requests"), targetRoot.resolve("requests")); - copyTreeIfExists(sourceRoot.resolve("policy"), targetRoot.resolve("policy")); - copyTreeIfExists(sourceRoot.resolve("publications"), targetRoot.resolve("publications")); - copyTreeIfExists(sourceRoot.resolve("sign-workflows"), targetRoot.resolve("sign-workflows")); - copyStagedContent(sourceRoot.resolve("staged-content"), targetRoot.resolve("staged-content"), - plan.authority().remintedContentIds(), plan.nonCredentialContentIds()); - copyImportedProfilesAsOf(plan.profiles(), targetRoot.resolve("profiles")); - reconstructMutableTree(sourceRoot.resolve("workflows"), targetRoot.resolve("workflows"), at, - options.workflowHistoryPolicy(), options.strictSnapshotExport()); - new AuthorityRestorer(source, options, plan.authority()).restore(targetRoot); - installRevocationAuthority(targetRoot, stagedRevocationPrefix); - try (FilesystemPkiStore ignored = new FilesystemPkiStore(targetRoot, options)) { - // Opening validates the restored authority and rebuilds its derived index. + try (PublicationCursor publications = source.openPublicationRecords()) { + Path sourceRoot = source.snapshotRoot(); + Path stagedRevocationPrefix = targetRoot.resolve(REVOCATION_PREFIX_STAGE); + FilesystemRevocationLog.RecoveryTarget captured = + source.copyRevocationPrefix(stagedRevocationPrefix); + copyFile(sourceRoot.resolve("SIGNING_NAMESPACE"), targetRoot.resolve("SIGNING_NAMESPACE")); + copyFile(sourceRoot.resolve("SIGNING_TIME_WATERMARK"), targetRoot.resolve("SIGNING_TIME_WATERMARK")); + copyTreeIfExists(sourceRoot.resolve("requests"), targetRoot.resolve("requests")); + copyTreeIfExists(sourceRoot.resolve("policy"), targetRoot.resolve("policy")); + copyTreeIfExists(sourceRoot.resolve("sign-workflows"), targetRoot.resolve("sign-workflows")); + copyStagedContent(sourceRoot.resolve("staged-content"), targetRoot.resolve("staged-content"), + plan.authority().remintedContentIds(), plan.nonCredentialContentIds()); + copyImportedProfilesAsOf(plan.profiles(), targetRoot.resolve("profiles")); + reconstructMutableTree(sourceRoot.resolve("workflows"), targetRoot.resolve("workflows"), at, + options.workflowHistoryPolicy(), options.strictSnapshotExport()); + new AuthorityRestorer(source, options, plan.authority()).restore(targetRoot); + installRevocationAuthority(targetRoot, stagedRevocationPrefix); + try (FilesystemPkiStore target = new FilesystemPkiStore(targetRoot, options)) { + // Opening validates the restored authority and rebuilds its derived index. + restorePublicationMetadata(publications, target); + } + writeSnapshotBoundary(targetRoot, captured.boundary()); + } + } + + @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops") + private static void restorePublicationMetadata(PublicationCursor cursor, FilesystemPkiStore target) + throws IOException { + Optional next; + while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) { + PublicationRecord record = next.orElseThrow(); + DurableContentReference targetContent = record.sourceType() == PublicationSourceType.CREDENTIAL + ? target.getCredential(record.sourceId()).map(Credential::content).orElseThrow( + () -> new IOException("Snapshot publication credential is missing")) + : target.getStatusObject(record.sourceId()).map(StatusObject::content).orElseThrow( + () -> new IOException("Snapshot publication status object is missing")); + PublicationRecord restored = new PublicationRecord(record.publicationId(), record.sourceType(), + record.sourceId(), targetContent, record.target(), record.status(), record.attemptNumber(), + record.attemptToken(), record.createdAt(), record.updatedAt(), record.failure(), + record.evidence()); + target.createPublicationRecord(restored); } - writeSnapshotBoundary(targetRoot, captured.boundary()); } private static void writeSnapshotBoundary(Path targetRoot, long boundary) throws IOException { diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/PublicationRecordCodec.java b/pki/src/main/java/zeroecho/pki/impl/fs/PublicationRecordCodec.java new file mode 100644 index 0000000..56d453e --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/impl/fs/PublicationRecordCodec.java @@ -0,0 +1,305 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.impl.fs; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Optional; +import zeroecho.pki.api.Encoding; +import zeroecho.pki.api.PkiId; +import zeroecho.pki.api.content.DurableContentReference; +import zeroecho.pki.api.publication.PublicationEvidence; +import zeroecho.pki.api.publication.PublicationFailure; +import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationSourceType; +import zeroecho.pki.api.publication.PublicationStatus; +import zeroecho.pki.api.publication.PublicationTarget; +import zeroecho.pki.api.publication.PublicationTargetType; +import zeroecho.pki.spi.store.StagedContentStore; + +/** Strict current-schema codec for finite publication control metadata. */ +@SuppressWarnings({ "PMD.CyclomaticComplexity", "PMD.AvoidCatchingGenericException", + "PMD.PreserveStackTrace" }) +final class PublicationRecordCodec { + private static final int MAGIC = 0x5a505542; + private static final int VERSION = 1; + private static final int MAXIMUM_RECORD_BYTES = 32 * 1024; + private static final int MAXIMUM_STRING_BYTES = 4096; + private static final int OPTIONAL_ABSENT = 0; + private static final int OPTIONAL_PRESENT = 1; + + private PublicationRecordCodec() { + } + + /* package */ static byte[] encode(PublicationRecord record) { + try { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(512); + try (DataOutputStream output = new DataOutputStream(buffer)) { + output.writeInt(MAGIC); + output.writeInt(VERSION); + writeString(output, record.publicationId().value()); + output.writeInt(sourceTypeCode(record.sourceType())); + writeString(output, record.sourceId().value()); + writeContent(output, record.content()); + output.writeInt(targetTypeCode(record.target().type())); + writeString(output, record.target().targetId()); + output.writeInt(statusCode(record.status())); + output.writeLong(record.attemptNumber()); + writeOptionalString(output, record.attemptToken()); + writeInstant(output, record.createdAt()); + writeInstant(output, record.updatedAt()); + output.writeInt(record.failure().map(PublicationRecordCodec::failureCode).orElse(0)); + output.writeInt(record.evidence().map(PublicationRecordCodec::evidenceCode).orElse(0)); + } + byte[] encoded = buffer.toByteArray(); + if (encoded.length > MAXIMUM_RECORD_BYTES) { + throw new IllegalArgumentException("Publication record exceeds its finite metadata limit"); + } + return encoded; + } catch (IOException impossible) { + throw new IllegalStateException("Publication record encoding failed", impossible); + } + } + + /* package */ static PublicationRecord decode(byte[] encoded, StagedContentStore stagedContent) + throws IOException { + if (encoded.length > MAXIMUM_RECORD_BYTES) { + throw malformed(); + } + try (ByteArrayInputStream bytes = new ByteArrayInputStream(encoded); + DataInputStream input = new DataInputStream(bytes)) { + if (input.readInt() != MAGIC || input.readInt() != VERSION) { + throw malformed(); + } + PkiId publicationId = new PkiId(readString(input)); + PublicationSourceType sourceType = sourceType(input.readInt()); + PkiId sourceId = new PkiId(readString(input)); + DurableContentReference content = readContent(input, stagedContent); + PublicationTarget target = new PublicationTarget(targetType(input.readInt()), readString(input)); + PublicationStatus status = status(input.readInt()); + long attemptNumber = input.readLong(); + Optional attemptToken = readOptionalString(input); + Instant createdAt = readInstant(input); + Instant updatedAt = readInstant(input); + Optional failure = optionalFailure(input.readInt()); + Optional evidence = optionalEvidence(input.readInt()); + if (bytes.available() != 0) { + throw malformed(); + } + return new PublicationRecord(publicationId, sourceType, sourceId, content, target, status, + attemptNumber, attemptToken, createdAt, updatedAt, failure, evidence); + } catch (EOFException | RuntimeException failure) { + throw malformed(); + } + } + + private static void writeContent(DataOutputStream output, DurableContentReference content) throws IOException { + writeString(output, content.storeId()); + writeString(output, content.contentId()); + output.writeInt(encodingCode(content.encoding())); + output.writeLong(content.length()); + writeString(output, content.sha256()); + output.writeInt(lifecycleCode(content.lifecycle())); + } + + private static DurableContentReference readContent(DataInputStream input, StagedContentStore stagedContent) + throws IOException { + return stagedContent.restoreReference(readString(input), readString(input), encoding(input.readInt()), + input.readLong(), readString(input), lifecycle(input.readInt())); + } + + private static void writeString(DataOutputStream output, String value) throws IOException { + byte[] encoded = value.getBytes(StandardCharsets.UTF_8); + if (encoded.length == 0 || encoded.length > MAXIMUM_STRING_BYTES) { + throw new IllegalArgumentException("Publication metadata string is not finite"); + } + output.writeInt(encoded.length); + output.write(encoded); + } + + private static String readString(DataInputStream input) throws IOException { + int length = input.readInt(); + if (length <= 0 || length > MAXIMUM_STRING_BYTES) { + throw malformed(); + } + byte[] value = input.readNBytes(length); + if (value.length != length) { + throw malformed(); + } + String decoded = new String(value, StandardCharsets.UTF_8); + if (!java.util.Arrays.equals(value, decoded.getBytes(StandardCharsets.UTF_8))) { + throw malformed(); + } + return decoded; + } + + private static void writeOptionalString(DataOutputStream output, Optional value) throws IOException { + output.writeByte(value.isPresent() ? OPTIONAL_PRESENT : OPTIONAL_ABSENT); + if (value.isPresent()) { + writeString(output, value.orElseThrow()); + } + } + + private static Optional readOptionalString(DataInputStream input) throws IOException { + int marker = input.readUnsignedByte(); + if (marker == OPTIONAL_ABSENT) { + return Optional.empty(); + } + if (marker == OPTIONAL_PRESENT) { + return Optional.of(readString(input)); + } + throw malformed(); + } + + private static void writeInstant(DataOutputStream output, Instant value) throws IOException { + output.writeLong(value.getEpochSecond()); + output.writeInt(value.getNano()); + } + + private static Instant readInstant(DataInputStream input) throws IOException { + return Instant.ofEpochSecond(input.readLong(), input.readInt()); + } + + private static int sourceTypeCode(PublicationSourceType value) { + return switch (value) { + case CREDENTIAL -> 1; + case STATUS_OBJECT -> 2; + }; + } + + private static PublicationSourceType sourceType(int code) throws IOException { + return switch (code) { + case 1 -> PublicationSourceType.CREDENTIAL; + case 2 -> PublicationSourceType.STATUS_OBJECT; + default -> throw malformed(); + }; + } + + private static int targetTypeCode(PublicationTargetType value) { + return switch (value) { + case FILESYSTEM -> 1; + case LDAP -> 2; + case HTTP -> 3; + case OBJECT_STORE -> 4; + case CUSTOM -> 5; + }; + } + + private static PublicationTargetType targetType(int code) throws IOException { + return switch (code) { + case 1 -> PublicationTargetType.FILESYSTEM; + case 2 -> PublicationTargetType.LDAP; + case 3 -> PublicationTargetType.HTTP; + case 4 -> PublicationTargetType.OBJECT_STORE; + case 5 -> PublicationTargetType.CUSTOM; + default -> throw malformed(); + }; + } + + private static int statusCode(PublicationStatus value) { + return switch (value) { + case PENDING -> 1; + case DISPATCH_PREPARED -> 2; + case SUCCEEDED -> 3; + case RETRYABLE_FAILURE -> 4; + case TERMINAL_FAILURE -> 5; + case OUTCOME_UNKNOWN -> 6; + }; + } + + private static PublicationStatus status(int code) throws IOException { + return switch (code) { + case 1 -> PublicationStatus.PENDING; + case 2 -> PublicationStatus.DISPATCH_PREPARED; + case 3 -> PublicationStatus.SUCCEEDED; + case 4 -> PublicationStatus.RETRYABLE_FAILURE; + case 5 -> PublicationStatus.TERMINAL_FAILURE; + case 6 -> PublicationStatus.OUTCOME_UNKNOWN; + default -> throw malformed(); + }; + } + + private static int failureCode(PublicationFailure value) { + return switch (value) { + case EXTERNAL_RETRYABLE -> 1; + case EXTERNAL_TERMINAL -> 2; + case EXTERNAL_OUTCOME_UNKNOWN -> 3; + case CONTENT_INVALID -> 4; + case RESULT_PERSISTENCE_UNKNOWN -> 5; + }; + } + + private static Optional optionalFailure(int code) throws IOException { + return switch (code) { + case 0 -> Optional.empty(); + case 1 -> Optional.of(PublicationFailure.EXTERNAL_RETRYABLE); + case 2 -> Optional.of(PublicationFailure.EXTERNAL_TERMINAL); + case 3 -> Optional.of(PublicationFailure.EXTERNAL_OUTCOME_UNKNOWN); + case 4 -> Optional.of(PublicationFailure.CONTENT_INVALID); + case 5 -> Optional.of(PublicationFailure.RESULT_PERSISTENCE_UNKNOWN); + default -> throw malformed(); + }; + } + + private static int evidenceCode(PublicationEvidence value) { + return switch (value) { + case DIRECT_RESPONSE -> 1; + case RECONCILIATION_RESPONSE -> 2; + }; + } + + private static Optional optionalEvidence(int code) throws IOException { + return switch (code) { + case 0 -> Optional.empty(); + case 1 -> Optional.of(PublicationEvidence.DIRECT_RESPONSE); + case 2 -> Optional.of(PublicationEvidence.RECONCILIATION_RESPONSE); + default -> throw malformed(); + }; + } + + private static int encodingCode(Encoding value) { + return switch (value) { + case DER -> 1; + case PEM -> 2; + case BINARY -> 3; + }; + } + + private static Encoding encoding(int code) throws IOException { + return switch (code) { + case 1 -> Encoding.DER; + case 2 -> Encoding.PEM; + case 3 -> Encoding.BINARY; + default -> throw malformed(); + }; + } + + private static int lifecycleCode(DurableContentReference.Lifecycle value) { + return switch (value) { + case OPERATION -> 1; + case PERSISTED -> 2; + case TEMPORARY -> 3; + }; + } + + private static DurableContentReference.Lifecycle lifecycle(int code) throws IOException { + return switch (code) { + case 1 -> DurableContentReference.Lifecycle.OPERATION; + case 2 -> DurableContentReference.Lifecycle.PERSISTED; + case 3 -> DurableContentReference.Lifecycle.TEMPORARY; + default -> throw malformed(); + }; + } + + private static IOException malformed() { + return new IOException("Publication record metadata is malformed"); + } +} diff --git a/pki/src/main/java/zeroecho/pki/spi/publish/PublicationAttempt.java b/pki/src/main/java/zeroecho/pki/spi/publish/PublicationAttempt.java new file mode 100644 index 0000000..194803b --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/spi/publish/PublicationAttempt.java @@ -0,0 +1,43 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.spi.publish; + +import java.util.Objects; +import zeroecho.pki.api.Encoding; +import zeroecho.pki.api.PkiId; +import zeroecho.pki.api.publication.PublicationSourceType; +import zeroecho.pki.api.publication.PublicationTarget; + +/** + * Immutable non-secret identity and content commitment for one external attempt. + * + * @param publicationId operation identity + * @param sourceType authoritative source type + * @param sourceId authoritative source identity + * @param target configured target identity + * @param attemptNumber positive attempt number + * @param attemptToken stable canonical idempotency token + * @param encoding payload encoding + * @param length exact payload length + * @param sha256 exact lowercase SHA-256 commitment + */ +public record PublicationAttempt(PkiId publicationId, PublicationSourceType sourceType, PkiId sourceId, + PublicationTarget target, long attemptNumber, String attemptToken, Encoding encoding, + long length, String sha256) { + /** Creates a validated immutable attempt descriptor. */ + public PublicationAttempt { + Objects.requireNonNull(publicationId, "publicationId"); + Objects.requireNonNull(sourceType, "sourceType"); + Objects.requireNonNull(sourceId, "sourceId"); + Objects.requireNonNull(target, "target"); + Objects.requireNonNull(attemptToken, "attemptToken"); + Objects.requireNonNull(encoding, "encoding"); + Objects.requireNonNull(sha256, "sha256"); + if (attemptNumber <= 0L || length < 0L || !attemptToken.matches("[0-9a-f]{64}") + || !sha256.matches("[0-9a-f]{64}")) { + throw new IllegalArgumentException("Publication attempt commitment is invalid"); + } + } +} diff --git a/pki/src/main/java/zeroecho/pki/spi/publish/PublicationOutcome.java b/pki/src/main/java/zeroecho/pki/spi/publish/PublicationOutcome.java new file mode 100644 index 0000000..994ee51 --- /dev/null +++ b/pki/src/main/java/zeroecho/pki/spi/publish/PublicationOutcome.java @@ -0,0 +1,15 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.spi.publish; + +/** Definitive outcomes a publisher may safely return for one exact attempt. */ +public enum PublicationOutcome { + /** The exact attempt completed successfully. */ + SUCCESS, + /** No unsafe duplicate effect exists and an explicit retry is permitted. */ + RETRYABLE_FAILURE, + /** The exact attempt definitively failed and must not be retried. */ + TERMINAL_FAILURE +} diff --git a/pki/src/main/java/zeroecho/pki/spi/publish/Publisher.java b/pki/src/main/java/zeroecho/pki/spi/publish/Publisher.java index bbf6b22..a745fff 100644 --- a/pki/src/main/java/zeroecho/pki/spi/publish/Publisher.java +++ b/pki/src/main/java/zeroecho/pki/spi/publish/Publisher.java @@ -33,22 +33,42 @@ ******************************************************************************/ package zeroecho.pki.spi.publish; +import java.util.Optional; import zeroecho.core.io.RepeatableContent; import zeroecho.pki.api.publication.PublicationTarget; /** * Publishes an encoded artifact to a configured publication target. */ -@SuppressWarnings("PMD.ImplicitFunctionalInterface") public interface Publisher { /** - * Publishes the given payload to the specified target. + * Returns the exact administrator-configured destination served by this instance. * - * @param target publication target - * @param payload payload to publish - * @throws IllegalArgumentException if inputs are null - * @throws RuntimeException if publishing fails + * @return immutable destination identity */ - void publish(PublicationTarget target, RepeatableContent payload); + PublicationTarget target(); + + /** + * Publishes the payload for one durably prepared attempt. + * + * @param attempt exact attempt and idempotency identity + * @param payload repeatable immutable payload; implementations must stream it + * @return definitive classified outcome + * @throws IllegalArgumentException if inputs are null + * @throws RuntimeException if the outcome cannot be established; callers treat + * every such exception as externally unknown and discard its details + */ + PublicationOutcome publish(PublicationAttempt attempt, RepeatableContent payload); + + /** + * Resolves an existing unknown attempt without creating a new external effect. + * + * @param attempt original exact attempt identity + * @return definitive outcome, or empty when reliable reconciliation is unsupported + * @throws RuntimeException if the lookup outcome itself is unknown + */ + default Optional reconcile(PublicationAttempt attempt) { + return Optional.empty(); + } } diff --git a/pki/src/main/java/zeroecho/pki/spi/publish/package-info.java b/pki/src/main/java/zeroecho/pki/spi/publish/package-info.java index fcab78c..4ceedb0 100644 --- a/pki/src/main/java/zeroecho/pki/spi/publish/package-info.java +++ b/pki/src/main/java/zeroecho/pki/spi/publish/package-info.java @@ -36,7 +36,9 @@ * *

* Publishers implement target-specific distribution (filesystem, LDAP, HTTP, - * object stores). + * object stores) for one core-issued attempt token. They return only closed safe + * outcomes and may optionally reconcile that exact attempt. Provider secrets and + * exception details never cross this boundary into publication metadata. *

*/ package zeroecho.pki.spi.publish; diff --git a/pki/src/main/java/zeroecho/pki/spi/store/PkiStore.java b/pki/src/main/java/zeroecho/pki/spi/store/PkiStore.java index 10b8b6c..db64f3c 100644 --- a/pki/src/main/java/zeroecho/pki/spi/store/PkiStore.java +++ b/pki/src/main/java/zeroecho/pki/spi/store/PkiStore.java @@ -45,6 +45,7 @@ import zeroecho.pki.api.policy.PolicyTrace; import zeroecho.pki.api.profile.ActiveCertificateProfile; import zeroecho.pki.api.profile.CertificateProfileRef; import zeroecho.pki.api.profile.ImportedCertificateProfileVersion; +import zeroecho.pki.api.publication.PublicationCursor; import zeroecho.pki.api.publication.PublicationRecord; import zeroecho.pki.api.request.ParsedCertificationRequest; import zeroecho.pki.api.revocation.RevocationCommand; @@ -61,6 +62,21 @@ import zeroecho.pki.api.status.StatusObject; * must not leak secrets through logging or exception messages. *

* + *

Publication recovery ownership

+ *

+ * Before an instance is exposed to callers, its implementation must hold + * exclusive publication-runtime ownership and validate the complete publication + * namespace in one stable view. Any abandoned + * {@link zeroecho.pki.api.publication.PublicationStatus#DISPATCH_PREPARED} + * operation must be durably changed to + * {@link zeroecho.pki.api.publication.PublicationStatus#OUTCOME_UNKNOWN} before + * exposure, without invoking a publisher. Once exposed, the store must not + * reclassify a prepared operation because it may represent an active dispatch in + * the owning runtime. A backend that supports simultaneous runtimes must provide + * durable session or lease fencing with equivalent safety instead of unconditional + * open-time recovery. + *

+ * *

Security

*

* Implementations must protect persisted data appropriately (for example: @@ -256,27 +272,49 @@ public interface PkiStore extends SignWorkflowStore { List listStatusObjects(PkiId issuerCaId); /** - * Persists or updates a publication record. + * Creates one publication operation if absent. * *

- * Publication records describe distribution state (for example: where and when - * an object was published). These records may be used for operational - * monitoring and reconciliation. + * An identical replay is idempotent. A different record under the same identity + * fails closed. The implementation must use its finite transactional metadata + * authority and must not create a sidecar record. *

* * @param record publication record (never {@code null}) * @throws NullPointerException if {@code record} is {@code null} - * @throws IllegalStateException if persistence fails + * @return committed record, including an identical existing record + * @throws IllegalArgumentException if the identity conflicts + * @throws IllegalStateException if persistence fails or is uncertain */ - void putPublicationRecord(PublicationRecord record); + PublicationRecord createPublicationRecord(PublicationRecord record); /** - * Lists all publication records. + * Reads one publication operation by exact semantic identity. * - * @return list of publication records (never {@code null}) - * @throws IllegalStateException if listing fails + * @param publicationId operation identity + * @return current durable record, or empty when absent */ - List listPublicationRecords(); + Optional getPublicationRecord(PkiId publicationId); + + /** + * Atomically replaces one exact observed publication state. + * + * @param expected exact current record + * @param replacement valid next state with identical immutable fields + * @return {@code true} when committed; {@code false} on a concurrent conflict + * @throws IllegalArgumentException if identities, immutable fields or the state + * transition are invalid + * @throws IllegalStateException if persistence fails or is uncertain + */ + boolean replacePublicationRecord(PublicationRecord expected, PublicationRecord replacement); + + /** + * Opens a stable ordered cursor over the complete publication namespace. + * + * @return closeable cursor whose heap does not grow with returned records + * @throws IllegalStateException if the stable view cannot be opened + */ + PublicationCursor openPublicationRecords(); /** * Atomically imports one immutable validated profile version. diff --git a/pki/src/main/java/zeroecho/pki/spi/store/PkiStoreProvider.java b/pki/src/main/java/zeroecho/pki/spi/store/PkiStoreProvider.java index 5780ce3..d8a11ca 100644 --- a/pki/src/main/java/zeroecho/pki/spi/store/PkiStoreProvider.java +++ b/pki/src/main/java/zeroecho/pki/spi/store/PkiStoreProvider.java @@ -64,6 +64,9 @@ public interface PkiStoreProvider extends ConfigurableProvider { *

* Implementations must validate required keys and throw * {@link IllegalArgumentException} if configuration is incomplete or invalid. + * Before returning, the provider must also satisfy the exclusive publication + * recovery contract documented by {@link PkiStore}; no external publication + * call may occur during allocation. *

* * @param config configuration (never {@code null}) diff --git a/pki/src/test/java/zeroecho/pki/e2e/PkiProofGateE2eTest.java b/pki/src/test/java/zeroecho/pki/e2e/PkiProofGateE2eTest.java index f30348c..920d923 100644 --- a/pki/src/test/java/zeroecho/pki/e2e/PkiProofGateE2eTest.java +++ b/pki/src/test/java/zeroecho/pki/e2e/PkiProofGateE2eTest.java @@ -611,7 +611,10 @@ final class PkiProofGateE2eTest { assertTrue(signingFailure.store().listCas().isEmpty()); assertTrue(signingFailure.store().listWorkflowStates().isEmpty()); assertTrue(signingFailure.store().listSignRecords().isEmpty()); - assertTrue(signingFailure.store().listPublicationRecords().isEmpty()); + try (zeroecho.pki.api.publication.PublicationCursor cursor = + signingFailure.store().openPublicationRecords()) { + assertTrue(cursor.next(zeroecho.core.io.CancellationSignal.NONE).isEmpty()); + } assertFalse(signingFailure.hasRunningSignatureOperations()); assertEquals(0, signingFailure.submittedSignCount()); assertEquals("MANAGED_KEY_UNAVAILABLE", signingFailure.auditSink().snapshot() diff --git a/pki/src/test/java/zeroecho/pki/impl/fs/DefaultPublicationServiceTest.java b/pki/src/test/java/zeroecho/pki/impl/fs/DefaultPublicationServiceTest.java new file mode 100644 index 0000000..8933efe --- /dev/null +++ b/pki/src/test/java/zeroecho/pki/impl/fs/DefaultPublicationServiceTest.java @@ -0,0 +1,624 @@ +/******************************************************************************* + * Copyright (C) 2026, Leo Galambos + * All rights reserved. + *******************************************************************************/ +package zeroecho.pki.impl.fs; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import zeroecho.core.io.CancellationSignal; +import zeroecho.core.io.RepeatableContent; +import zeroecho.pki.api.Encoding; +import zeroecho.pki.api.FormatId; +import zeroecho.pki.api.IssuerRef; +import zeroecho.pki.api.PkiException; +import zeroecho.pki.api.PkiId; +import zeroecho.pki.api.SubjectRef; +import zeroecho.pki.api.Validity; +import zeroecho.pki.api.content.DurableContentReference; +import zeroecho.pki.api.credential.CaProfileBinding; +import zeroecho.pki.api.credential.Credential; +import zeroecho.pki.api.credential.CredentialStatus; +import zeroecho.pki.api.publication.PublicationCursor; +import zeroecho.pki.api.publication.PublicationFailure; +import zeroecho.pki.api.publication.PublicationQuery; +import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationRequest; +import zeroecho.pki.api.publication.PublicationResult; +import zeroecho.pki.api.publication.PublicationSourceType; +import zeroecho.pki.api.publication.PublicationStatus; +import zeroecho.pki.api.publication.PublicationTarget; +import zeroecho.pki.api.publication.PublicationTargetType; +import zeroecho.pki.api.status.StatusObject; +import zeroecho.pki.api.status.StatusObjectType; +import zeroecho.pki.impl.core.DefaultPublicationService; +import zeroecho.pki.impl.core.attr.SimpleAttributeSet; +import zeroecho.pki.spi.publish.PublicationAttempt; +import zeroecho.pki.spi.publish.PublicationOutcome; +import zeroecho.pki.spi.publish.Publisher; + +final class DefaultPublicationServiceTest { + private static final Instant NOW = Instant.parse("2026-08-03T10:15:30Z"); + private static final Clock CLOCK = Clock.fixed(NOW, ZoneOffset.UTC); + private static final PublicationTarget TARGET = + new PublicationTarget(PublicationTargetType.FILESYSTEM, "primary-mirror-v1"); + + @TempDir + private Path temporaryDirectory; + + @AfterEach + void clearFaults() { + FilesystemPkiStore.clearPublicationCommitFault(); + } + + @Test + void registrationRequiresCommittedExactSourceAndIsIdempotent() throws Exception { + System.out.println("registrationRequiresCommittedExactSourceAndIsIdempotent"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore store = store("registration"); + DefaultPublicationService service = service(store, publisher)) { + Credential credential = credential(store, "registration-one", new byte[] { 1, 2, 3 }); + PublicationRequest request = credentialRequest("publication-registration", credential); + assertThrows(IllegalArgumentException.class, () -> service.register(request)); + assertEquals(0, publisher.dispatchCalls.get()); + + store.putCredential(credential); + Set ownersBefore = + store.stagedContent().contentOwners(credential.content()); + PublicationRecord first = service.register(request); + PublicationRecord replay = service.register(request); + assertEquals(PublicationStatus.PENDING, first.status()); + assertEquals(first, replay); + assertEquals(ownersBefore, store.stagedContent().contentOwners(credential.content())); + assertEquals(0, publisher.dispatchCalls.get()); + + Credential other = credential(store, "registration-two", new byte[] { 4, 5, 6 }); + store.putCredential(other); + PublicationRequest conflict = new PublicationRequest(first.publicationId(), + PublicationSourceType.CREDENTIAL, other.credentialId(), TARGET); + assertThrows(IllegalArgumentException.class, () -> service.register(conflict)); + PublicationRecord changedTarget = new PublicationRecord(first.publicationId(), first.sourceType(), + first.sourceId(), first.content(), + new PublicationTarget(PublicationTargetType.FILESYSTEM, "other-mirror-v1"), first.status(), + first.attemptNumber(), first.attemptToken(), first.createdAt(), first.updatedAt(), + first.failure(), first.evidence()); + assertThrows(IllegalArgumentException.class, + () -> store.replacePublicationRecord(first, changedTarget)); + + StatusObject status = status(store, "registration-status", new byte[] { 7, 8 }); + store.putStatusObject(status); + PublicationRecord statusRecord = service.register(statusRequest("publication-status", status)); + assertEquals(status.content(), statusRecord.content()); + assertEquals(PublicationSourceType.STATUS_OBJECT, statusRecord.sourceType()); + + PublicationTarget absent = new PublicationTarget(PublicationTargetType.HTTP, "unconfigured-v1"); + assertThrows(IllegalArgumentException.class, () -> service.register(new PublicationRequest( + new PkiId("publication-absent"), PublicationSourceType.CREDENTIAL, + credential.credentialId(), absent))); + System.out.println("...registered states=" + first.status() + "," + statusRecord.status()); + } + System.out.println("registrationRequiresCommittedExactSourceAndIsIdempotent...ok"); + } + + @Test + void storeRejectsForeignContentAndObsoleteSidecarAuthority() throws Exception { + System.out.println("storeRejectsForeignContentAndObsoleteSidecarAuthority"); + Path firstRoot = temporaryDirectory.resolve("foreign-first"); + Path secondRoot = temporaryDirectory.resolve("foreign-second"); + DurableContentReference foreign; + try (FilesystemPkiStore second = new FilesystemPkiStore(secondRoot, FsPkiStoreOptions.defaults())) { + foreign = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(second.stagedContent(), Encoding.DER, + new byte[] { 9, 9, 9 }); + } + try (FilesystemPkiStore first = new FilesystemPkiStore(firstRoot, FsPkiStoreOptions.defaults())) { + Credential credential = credential(first, "foreign", new byte[] { 1 }); + first.putCredential(credential); + PublicationRecord invalid = pending("publication-foreign", credential, foreign); + assertThrows(IllegalArgumentException.class, () -> first.createPublicationRecord(invalid)); + } + Path obsolete = firstRoot.resolve("publications").resolve("by-id").resolve("legacy.bin"); + Files.createDirectories(obsolete.getParent()); + Files.write(obsolete, new byte[] { 1, 2 }); + assertThrows(IllegalStateException.class, + () -> new FilesystemPkiStore(firstRoot, FsPkiStoreOptions.defaults())); + assertArrayEquals(new byte[] { 1, 2 }, Files.readAllBytes(obsolete)); + System.out.println("storeRejectsForeignContentAndObsoleteSidecarAuthority...ok"); + } + + @Test + void dispatchRetryAndTerminalStatesAreDurable() throws Exception { + System.out.println("dispatchRetryAndTerminalStatesAreDurable"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore store = store("dispatch-states"); + DefaultPublicationService service = service(store, publisher)) { + Credential credential = committedCredential(store, "retry", new byte[] { 1, 2, 3, 4 }); + PkiId operation = service.register(credentialRequest("publication-retry", credential)).publicationId(); + publisher.dispatchOutcome.set(PublicationOutcome.RETRYABLE_FAILURE); + PublicationResult retryable = service.process(operation); + PublicationRecord firstAttempt = service.find(operation).orElseThrow(); + String firstToken = firstAttempt.attemptToken().orElseThrow(); + assertEquals(PublicationStatus.RETRYABLE_FAILURE, retryable.status()); + assertEquals(1L, retryable.attemptNumber()); + + publisher.dispatchOutcome.set(PublicationOutcome.SUCCESS); + PublicationResult success = service.retry(operation); + PublicationRecord secondAttempt = service.find(operation).orElseThrow(); + assertEquals(PublicationStatus.SUCCEEDED, success.status()); + assertEquals(2L, success.attemptNumber()); + assertNotEquals(firstToken, secondAttempt.attemptToken().orElseThrow()); + assertThrows(IllegalArgumentException.class, () -> service.retry(operation)); + assertThrows(IllegalArgumentException.class, () -> service.process(operation)); + + Credential terminalCredential = committedCredential(store, "terminal", new byte[] { 5, 6 }); + PkiId terminalId = service.register( + credentialRequest("publication-terminal", terminalCredential)).publicationId(); + publisher.dispatchOutcome.set(PublicationOutcome.TERMINAL_FAILURE); + assertEquals(PublicationStatus.TERMINAL_FAILURE, service.process(terminalId).status()); + assertThrows(IllegalArgumentException.class, () -> service.retry(terminalId)); + assertEquals(3, publisher.dispatchCalls.get()); + assertTrue(publisher.maximumReadSize.get() <= 11); + System.out.println("...attempts=" + secondAttempt.attemptNumber()); + } + System.out.println("dispatchRetryAndTerminalStatesAreDurable...ok"); + } + + @Test + void publisherExceptionIsRedactedUnknownAndReconcilesWithSameToken() throws Exception { + System.out.println("publisherExceptionIsRedactedUnknownAndReconcilesWithSameToken"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore store = store("unknown"); + DefaultPublicationService service = service(store, publisher)) { + Credential credential = committedCredential(store, "unknown", new byte[] { 1, 3, 5 }); + PkiId operation = service.register(credentialRequest("publication-unknown", credential)).publicationId(); + publisher.throwOnDispatch.set(true); + PublicationResult unknown = service.process(operation); + PublicationRecord unknownRecord = service.find(operation).orElseThrow(); + String token = unknownRecord.attemptToken().orElseThrow(); + assertEquals(PublicationStatus.OUTCOME_UNKNOWN, unknown.status()); + assertEquals(Optional.of(PublicationFailure.EXTERNAL_OUTCOME_UNKNOWN), unknownRecord.failure()); + assertThrows(IllegalArgumentException.class, () -> service.process(operation)); + + publisher.throwOnDispatch.set(false); + PublicationResult unresolved = service.reconcile(operation); + assertEquals(PublicationStatus.OUTCOME_UNKNOWN, unresolved.status()); + assertEquals(token, service.find(operation).orElseThrow().attemptToken().orElseThrow()); + + publisher.reconciliationOutcome.set(Optional.of(PublicationOutcome.SUCCESS)); + PublicationResult resolved = service.reconcile(operation); + assertEquals(PublicationStatus.SUCCEEDED, resolved.status()); + assertEquals(token, publisher.reconciliationAttempts.get(1).attemptToken()); + assertEquals(1, publisher.dispatchCalls.get()); + System.out.println("...token=" + token.substring(0, 12) + "..."); + } + System.out.println("publisherExceptionIsRedactedUnknownAndReconcilesWithSameToken...ok"); + } + + @Test + void resultPersistenceFailureFencesUnknownAndCommittedUncertaintyResolves() throws Exception { + System.out.println("resultPersistenceFailureFencesUnknownAndCommittedUncertaintyResolves"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore store = store("persistence-failure"); + DefaultPublicationService service = service(store, publisher)) { + Credential first = committedCredential(store, "persist-first", new byte[] { 1 }); + PkiId firstId = service.register(credentialRequest("publication-persist-first", first)).publicationId(); + publisher.afterDispatch = () -> FilesystemPkiStore.installPublicationCommitFault( + FilesystemPkiStore.PublicationCommitFaultPoint.REPLACE_BEFORE_COMMIT); + PkiException failure = assertThrows(PkiException.class, () -> service.process(firstId)); + assertTrue(failure.getMessage().contains("code=PUBLICATION_RESULT_UNCONFIRMED")); + PublicationRecord unknown = service.find(firstId).orElseThrow(); + assertEquals(PublicationStatus.OUTCOME_UNKNOWN, unknown.status()); + assertEquals(Optional.of(PublicationFailure.RESULT_PERSISTENCE_UNKNOWN), unknown.failure()); + + Credential second = committedCredential(store, "persist-second", new byte[] { 2 }); + PkiId secondId = service.register( + credentialRequest("publication-persist-second", second)).publicationId(); + publisher.afterDispatch = () -> FilesystemPkiStore.installPublicationCommitFault( + FilesystemPkiStore.PublicationCommitFaultPoint.REPLACE_AFTER_COMMIT_AS_UNKNOWN); + PublicationResult resolved = service.process(secondId); + assertEquals(PublicationStatus.SUCCEEDED, resolved.status()); + assertEquals(PublicationStatus.SUCCEEDED, service.find(secondId).orElseThrow().status()); + + Credential third = committedCredential(store, "persist-third", new byte[] { 3 }); + FilesystemPkiStore.installPublicationCommitFault( + FilesystemPkiStore.PublicationCommitFaultPoint.CREATE_BEFORE_COMMIT); + assertThrows(IllegalStateException.class, + () -> service.register(credentialRequest("publication-create-failure", third))); + assertTrue(service.find(new PkiId("publication-create-failure")).isEmpty()); + System.out.println("...unknown failure=" + unknown.failure().orElseThrow()); + } + System.out.println("resultPersistenceFailureFencesUnknownAndCommittedUncertaintyResolves...ok"); + } + + @Test + void restartConvertsPreparedAttemptToUnknownWithoutDispatch() throws Exception { + System.out.println("restartConvertsPreparedAttemptToUnknownWithoutDispatch"); + Path root = temporaryDirectory.resolve("restart"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + PkiId operation = new PkiId("publication-restart"); + try (FilesystemPkiStore store = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults()); + DefaultPublicationService service = service(store, publisher)) { + Credential credential = committedCredential(store, "restart", new byte[] { 4, 4 }); + PublicationRecord pending = service.register(new PublicationRequest(operation, + PublicationSourceType.CREDENTIAL, credential.credentialId(), TARGET)); + PublicationRecord prepared = new PublicationRecord(pending.publicationId(), pending.sourceType(), + pending.sourceId(), pending.content(), pending.target(), PublicationStatus.DISPATCH_PREPARED, + 1L, Optional.of("a".repeat(64)), pending.createdAt(), pending.updatedAt(), Optional.empty(), + Optional.empty()); + assertTrue(store.replacePublicationRecord(pending, prepared)); + } + try (FilesystemPkiStore reopened = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults())) { + PublicationRecord record = reopened.getPublicationRecord(operation).orElseThrow(); + assertEquals(PublicationStatus.OUTCOME_UNKNOWN, record.status()); + assertEquals("a".repeat(64), record.attemptToken().orElseThrow()); + assertEquals(0, publisher.dispatchCalls.get()); + try (DefaultPublicationService recovered = service(reopened, publisher)) { + assertEquals(PublicationStatus.OUTCOME_UNKNOWN, + recovered.find(operation).orElseThrow().status()); + } + } + System.out.println("restartConvertsPreparedAttemptToUnknownWithoutDispatch...ok"); + } + + @Test + void restartPreservesPendingRetryableSuccessfulAndUnknownStates() throws Exception { + System.out.println("restartPreservesPendingRetryableSuccessfulAndUnknownStates"); + Path root = temporaryDirectory.resolve("restart-states"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore store = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults()); + DefaultPublicationService service = service(store, publisher)) { + Credential pending = committedCredential(store, "restart-pending", new byte[] { 1 }); + Credential retryable = committedCredential(store, "restart-retryable", new byte[] { 2 }); + Credential succeeded = committedCredential(store, "restart-succeeded", new byte[] { 3 }); + Credential unknown = committedCredential(store, "restart-unknown", new byte[] { 4 }); + service.register(credentialRequest("publication-restart-pending", pending)); + service.register(credentialRequest("publication-restart-retryable", retryable)); + service.register(credentialRequest("publication-restart-succeeded", succeeded)); + service.register(credentialRequest("publication-restart-unknown", unknown)); + publisher.dispatchOutcome.set(PublicationOutcome.RETRYABLE_FAILURE); + service.process(new PkiId("publication-restart-retryable")); + publisher.dispatchOutcome.set(PublicationOutcome.SUCCESS); + service.process(new PkiId("publication-restart-succeeded")); + publisher.throwOnDispatch.set(true); + service.process(new PkiId("publication-restart-unknown")); + } + RecordingPublisher reopenedPublisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore reopened = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults()); + DefaultPublicationService recovered = service(reopened, reopenedPublisher)) { + assertEquals(PublicationStatus.PENDING, + recovered.find(new PkiId("publication-restart-pending")).orElseThrow().status()); + assertEquals(PublicationStatus.RETRYABLE_FAILURE, + recovered.find(new PkiId("publication-restart-retryable")).orElseThrow().status()); + assertEquals(PublicationStatus.SUCCEEDED, + recovered.find(new PkiId("publication-restart-succeeded")).orElseThrow().status()); + assertEquals(PublicationStatus.OUTCOME_UNKNOWN, + recovered.find(new PkiId("publication-restart-unknown")).orElseThrow().status()); + assertEquals(0, reopenedPublisher.dispatchCalls.get()); + assertTrue(reopened.getCredential(new PkiId("credential:restart-unknown")).isPresent()); + System.out.println("...recovered states=4"); + } + System.out.println("restartPreservesPendingRetryableSuccessfulAndUnknownStates...ok"); + } + + @Test + void concurrentSameOperationDoesNotDuplicateAndUnrelatedProgresses() throws Exception { + System.out.println("concurrentSameOperationDoesNotDuplicateAndUnrelatedProgresses"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + publisher.blockedPublication.set(new PkiId("publication-blocked")); + try (FilesystemPkiStore store = store("concurrency"); + DefaultPublicationService service = service(store, publisher); + ExecutorService executor = Executors.newFixedThreadPool(2)) { + Credential blocked = committedCredential(store, "blocked", new byte[] { 1, 2 }); + Credential unrelated = committedCredential(store, "unrelated", new byte[] { 3, 4 }); + service.register(credentialRequest("publication-blocked", blocked)); + service.register(credentialRequest("publication-unrelated", unrelated)); + + Future first = executor.submit( + () -> service.process(new PkiId("publication-blocked"))); + assertTrue(publisher.entered.await(5, TimeUnit.SECONDS)); + assertThrows(IllegalArgumentException.class, + () -> service.process(new PkiId("publication-blocked"))); + Future second = executor.submit( + () -> service.process(new PkiId("publication-unrelated"))); + assertEquals(PublicationStatus.SUCCEEDED, second.get(5, TimeUnit.SECONDS).status()); + service.close(); + assertThrows(IllegalStateException.class, + () -> service.find(new PkiId("publication-unrelated"))); + publisher.release.countDown(); + assertEquals(PublicationStatus.SUCCEEDED, first.get(5, TimeUnit.SECONDS).status()); + assertEquals(2, publisher.dispatchCalls.get()); + System.out.println("...dispatch calls=" + publisher.dispatchCalls.get()); + } + System.out.println("concurrentSameOperationDoesNotDuplicateAndUnrelatedProgresses...ok"); + } + + @Test + void secondServiceDoesNotReclassifyAnActiveDispatch() throws Exception { + System.out.println("secondServiceDoesNotReclassifyAnActiveDispatch"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + PkiId operation = new PkiId("publication-two-services"); + publisher.blockedPublication.set(operation); + try (FilesystemPkiStore store = store("two-services"); + DefaultPublicationService first = service(store, publisher); + ExecutorService executor = Executors.newSingleThreadExecutor()) { + Credential credential = committedCredential(store, "two-services", new byte[] { 7, 7 }); + first.register(new PublicationRequest(operation, PublicationSourceType.CREDENTIAL, + credential.credentialId(), TARGET)); + Future result = executor.submit(() -> first.process(operation)); + assertTrue(publisher.entered.await(5L, TimeUnit.SECONDS)); + try (DefaultPublicationService second = service(store, publisher)) { + assertEquals(PublicationStatus.DISPATCH_PREPARED, + second.find(operation).orElseThrow().status()); + assertThrows(IllegalArgumentException.class, () -> second.reconcile(operation)); + } + publisher.release.countDown(); + assertEquals(PublicationStatus.SUCCEEDED, result.get(5L, TimeUnit.SECONDS).status()); + assertEquals(1, publisher.dispatchCalls.get()); + System.out.println("...active state preserved"); + } + System.out.println("secondServiceDoesNotReclassifyAnActiveDispatch...ok"); + } + + @Test + void cursorIsStableFilteredAndCloseable() throws Exception { + System.out.println("cursorIsStableFilteredAndCloseable"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore store = store("cursor"); + DefaultPublicationService service = service(store, publisher)) { + Credential credential = committedCredential(store, "cursor-credential", new byte[] { 1 }); + StatusObject status = status(store, "cursor-status", new byte[] { 2 }); + store.putStatusObject(status); + service.register(credentialRequest("publication-cursor-credential", credential)); + PublicationCursor cursor = service.openPublications(new PublicationQuery(Optional.empty(), + Optional.empty(), Optional.empty(), Optional.of(PublicationSourceType.STATUS_OBJECT))); + service.register(statusRequest("publication-cursor-status", status)); + assertTrue(cursor.next(CancellationSignal.NONE).isEmpty()); + cursor.close(); + assertThrows(IllegalStateException.class, () -> cursor.next(CancellationSignal.NONE)); + + try (PublicationCursor current = service.openPublications(new PublicationQuery(Optional.empty(), + Optional.empty(), Optional.empty(), Optional.of(PublicationSourceType.STATUS_OBJECT)))) { + assertEquals(new PkiId("publication-cursor-status"), + current.next(CancellationSignal.NONE).orElseThrow().publicationId()); + assertTrue(current.next(CancellationSignal.NONE).isEmpty()); + } + } + System.out.println("cursorIsStableFilteredAndCloseable...ok"); + } + + @Test + void snapshotReconstructsPublicationMetadataWithoutSidecarAuthority() throws Exception { + System.out.println("snapshotReconstructsPublicationMetadataWithoutSidecarAuthority"); + Path sourceRoot = temporaryDirectory.resolve("snapshot-source"); + Path snapshotRoot = temporaryDirectory.resolve("snapshot-target"); + RecordingPublisher publisher = new RecordingPublisher(TARGET); + PkiId operation = new PkiId("publication-snapshot"); + String sourceContentId; + try (FilesystemPkiStore source = new FilesystemPkiStore(sourceRoot, FsPkiStoreOptions.defaults()); + DefaultPublicationService service = service(source, publisher)) { + Credential credential = committedCredential(source, "snapshot", new byte[] { 8, 6, 7, 5 }); + sourceContentId = credential.content().contentId(); + service.register(new PublicationRequest(operation, PublicationSourceType.CREDENTIAL, + credential.credentialId(), TARGET)); + assertEquals(PublicationStatus.SUCCEEDED, service.process(operation).status()); + source.exportSnapshot(snapshotRoot, NOW.plusSeconds(1L)); + } + assertFalse(Files.exists(snapshotRoot.resolve("publications"))); + try (FilesystemPkiStore restored = new FilesystemPkiStore(snapshotRoot, FsPkiStoreOptions.defaults())) { + PublicationRecord record = restored.getPublicationRecord(operation).orElseThrow(); + assertEquals(PublicationStatus.SUCCEEDED, record.status()); + assertNotEquals(sourceContentId, record.content().contentId()); + assertEquals(restored.getCredential(record.sourceId()).orElseThrow().content(), record.content()); + } + System.out.println("snapshotReconstructsPublicationMetadataWithoutSidecarAuthority...ok"); + } + + @Test + void snapshotRemintingPreservesPendingAndRetryAttemptTokens() throws Exception { + System.out.println("snapshotRemintingPreservesPendingAndRetryAttemptTokens"); + Path sourceRoot = temporaryDirectory.resolve("snapshot-token-source"); + Path snapshotRoot = temporaryDirectory.resolve("snapshot-token-target"); + PkiId pendingId = new PkiId("publication-snapshot-pending-token"); + PkiId retryId = new PkiId("publication-snapshot-retry-token"); + RecordingPublisher sourcePublisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore source = new FilesystemPkiStore(sourceRoot, FsPkiStoreOptions.defaults()); + DefaultPublicationService service = service(source, sourcePublisher)) { + Credential pending = committedCredential(source, "snapshot-pending-token", new byte[] { 1, 4 }); + Credential retryable = committedCredential(source, "snapshot-retry-token", new byte[] { 2, 8 }); + service.register(new PublicationRequest(pendingId, PublicationSourceType.CREDENTIAL, + pending.credentialId(), TARGET)); + service.register(new PublicationRequest(retryId, PublicationSourceType.CREDENTIAL, + retryable.credentialId(), TARGET)); + sourcePublisher.dispatchOutcome.set(PublicationOutcome.RETRYABLE_FAILURE); + service.process(retryId); + source.exportSnapshot(snapshotRoot, NOW.plusSeconds(1L)); + + sourcePublisher.dispatchOutcome.set(PublicationOutcome.SUCCESS); + service.process(pendingId); + service.retry(retryId); + } + + RecordingPublisher restoredPublisher = new RecordingPublisher(TARGET); + try (FilesystemPkiStore restored = new FilesystemPkiStore(snapshotRoot, FsPkiStoreOptions.defaults()); + DefaultPublicationService service = service(restored, restoredPublisher)) { + service.process(pendingId); + service.retry(retryId); + } + assertEquals(attemptToken(sourcePublisher, pendingId, 1L), + attemptToken(restoredPublisher, pendingId, 1L)); + assertEquals(attemptToken(sourcePublisher, retryId, 2L), + attemptToken(restoredPublisher, retryId, 2L)); + System.out.println("...snapshot-stable tokens=2"); + System.out.println("snapshotRemintingPreservesPendingAndRetryAttemptTokens...ok"); + } + + @Test + void strictCodecRejectsTrailingAndImpossiblePublicationMetadata() throws Exception { + System.out.println("strictCodecRejectsTrailingAndImpossiblePublicationMetadata"); + try (FilesystemPkiStore store = store("codec")) { + Credential credential = committedCredential(store, "codec", new byte[] { 1, 2 }); + PublicationRecord record = pending("publication-codec", credential, credential.content()); + byte[] encoded = PublicationRecordCodec.encode(record); + byte[] trailing = java.util.Arrays.copyOf(encoded, encoded.length + 1); + assertThrows(IOException.class, () -> PublicationRecordCodec.decode(trailing, store.stagedContent())); + encoded[0] ^= 0x01; + assertThrows(IOException.class, () -> PublicationRecordCodec.decode(encoded, store.stagedContent())); + } + System.out.println("strictCodecRejectsTrailingAndImpossiblePublicationMetadata...ok"); + } + + private FilesystemPkiStore store(String suffix) { + return new FilesystemPkiStore(temporaryDirectory.resolve(suffix), FsPkiStoreOptions.defaults()); + } + + private static DefaultPublicationService service(FilesystemPkiStore store, Publisher publisher) { + return new DefaultPublicationService(store, CLOCK, List.of(publisher)); + } + + private static Credential committedCredential(FilesystemPkiStore store, String suffix, byte[] bytes) + throws IOException { + Credential credential = credential(store, suffix, bytes); + store.putCredential(credential); + return credential; + } + + private static Credential credential(FilesystemPkiStore store, String suffix, byte[] bytes) throws IOException { + DurableContentReference content = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference( + store.stagedContent(), Encoding.DER, bytes); + return new Credential(new PkiId("credential:" + suffix), new FormatId("x509"), + new IssuerRef(new PkiId("ca:publication")), new SubjectRef("CN=" + suffix), + new Validity(NOW.minusSeconds(60L), NOW.plusSeconds(3600L)), suffix, + new PkiId("key:" + suffix), + new CaProfileBinding(new zeroecho.pki.api.profile.CertificateProfileRef( + "default", 1L, new byte[32])), CredentialStatus.ISSUED, content, new SimpleAttributeSet()); + } + + private static StatusObject status(FilesystemPkiStore store, String suffix, byte[] bytes) throws IOException { + DurableContentReference content = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference( + store.stagedContent(), Encoding.DER, bytes); + return new StatusObject(new PkiId("status:" + suffix), new FormatId("x509"), + new PkiId("ca:publication"), StatusObjectType.CRL, NOW, Optional.of(NOW.plusSeconds(60L)), + content, new SimpleAttributeSet()); + } + + private static PublicationRequest credentialRequest(String publicationId, Credential credential) { + return new PublicationRequest(new PkiId(publicationId), PublicationSourceType.CREDENTIAL, + credential.credentialId(), TARGET); + } + + private static PublicationRequest statusRequest(String publicationId, StatusObject status) { + return new PublicationRequest(new PkiId(publicationId), PublicationSourceType.STATUS_OBJECT, + status.statusObjectId(), TARGET); + } + + private static PublicationRecord pending(String publicationId, Credential credential, + DurableContentReference content) { + return new PublicationRecord(new PkiId(publicationId), PublicationSourceType.CREDENTIAL, + credential.credentialId(), content, TARGET, PublicationStatus.PENDING, 0L, Optional.empty(), + NOW, NOW, Optional.empty(), Optional.empty()); + } + + private static String attemptToken(RecordingPublisher publisher, PkiId publicationId, long attemptNumber) { + return publisher.dispatchAttempts.stream() + .filter(attempt -> attempt.publicationId().equals(publicationId) + && attempt.attemptNumber() == attemptNumber) + .findFirst().orElseThrow().attemptToken(); + } + + /** Deterministic streaming publisher with controllable outcomes and barriers. */ + private static final class RecordingPublisher implements Publisher { + private final PublicationTarget target; + private final AtomicInteger dispatchCalls = new AtomicInteger(); + private final AtomicInteger maximumReadSize = new AtomicInteger(); + private final AtomicReference dispatchOutcome = + new AtomicReference<>(PublicationOutcome.SUCCESS); + private final AtomicReference> reconciliationOutcome = + new AtomicReference<>(Optional.empty()); + private final AtomicBoolean throwOnDispatch = new AtomicBoolean(); + private final List reconciliationAttempts = java.util.Collections.synchronizedList( + new ArrayList<>()); + private final List dispatchAttempts = java.util.Collections.synchronizedList( + new ArrayList<>()); + private final AtomicReference blockedPublication = new AtomicReference<>(); + private final CountDownLatch entered = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private volatile Runnable afterDispatch = () -> { }; + + private RecordingPublisher(PublicationTarget target) { + this.target = target; + } + + @Override + public PublicationTarget target() { + return target; + } + + @Override + public PublicationOutcome publish(PublicationAttempt attempt, RepeatableContent payload) { + dispatchCalls.incrementAndGet(); + dispatchAttempts.add(attempt); + try (InputStream input = payload.openStream()) { + byte[] buffer = new byte[11]; + while (true) { + int count = input.read(buffer); + if (count < 0) { + break; + } + maximumReadSize.accumulateAndGet(count, Math::max); + } + if (attempt.publicationId().equals(blockedPublication.get())) { + entered.countDown(); + if (!release.await(5L, TimeUnit.SECONDS)) { + throw new IllegalStateException("publisher barrier timeout"); + } + } + } catch (IOException exception) { + throw new IllegalStateException("provider-secret-path", exception); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("provider-secret-interrupted", exception); + } + if (throwOnDispatch.get()) { + throw new IllegalStateException("provider-secret-token-and-path"); + } + afterDispatch.run(); + afterDispatch = () -> { }; + return dispatchOutcome.get(); + } + + @Override + public Optional reconcile(PublicationAttempt attempt) { + reconciliationAttempts.add(attempt); + return reconciliationOutcome.get(); + } + } +} diff --git a/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemPkiStoreTest.java b/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemPkiStoreTest.java index 96f1f5a..9ccf09a 100644 --- a/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemPkiStoreTest.java +++ b/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemPkiStoreTest.java @@ -111,6 +111,7 @@ import zeroecho.pki.api.profile.SubjectPolicy; import zeroecho.pki.api.profile.SubjectRdnRule; import zeroecho.pki.api.profile.SubjectRdnType; import zeroecho.pki.api.publication.PublicationRecord; +import zeroecho.pki.api.publication.PublicationSourceType; import zeroecho.pki.api.publication.PublicationStatus; import zeroecho.pki.api.publication.PublicationTarget; import zeroecho.pki.api.publication.PublicationTargetType; @@ -175,16 +176,17 @@ public final class FilesystemPkiStoreTest { zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(store.stagedContent(), Encoding.DER, new byte[] { 1, 2 }), attributes); - PublicationRecord publication = new PublicationRecord(new PkiId("publication-all"), now, - new PublicationTarget(PublicationTargetType.FILESYSTEM, "target-all", attributes), - credential.credentialId(), "CREDENTIAL", PublicationStatus.PUBLISHED); + PublicationRecord publication = new PublicationRecord(new PkiId("publication-all"), + PublicationSourceType.CREDENTIAL, credential.credentialId(), credential.content(), + new PublicationTarget(PublicationTargetType.FILESYSTEM, "target-all"), + PublicationStatus.PENDING, 0L, Optional.empty(), now, now, Optional.empty(), Optional.empty()); store.putCa(ca); store.putCredential(credential); store.putRequest(request); RevocationRecord revocation = store.transitionRevocation(new RevocationCommand.RevokePermanently( credential.credentialId(), RevocationReason.KEY_COMPROMISE, attributes), now); store.putStatusObject(status); - store.putPublicationRecord(publication); + store.createPublicationRecord(publication); importProfile(store, profile, now); store.activateProfile(profile.profileId(), 1); store.putPolicyTrace(trace); @@ -198,7 +200,8 @@ public final class FilesystemPkiStoreTest { store.getRevocation(revocation.credentialId()).orElseThrow().credentialId()); assertEquals(status.statusObjectId(), store.getStatusObject(status.statusObjectId()).orElseThrow().statusObjectId()); - assertEquals(publication.publicationId(), store.listPublicationRecords().get(0).publicationId()); + assertEquals(publication.publicationId(), + store.getPublicationRecord(publication.publicationId()).orElseThrow().publicationId()); assertEquals(profile, store.requireActiveProfile(profile.profileId()).profile()); assertEquals(trace, store.getPolicyTrace(trace.decisionId()).orElseThrow()); assertEquals(workflow.opId(), store.getWorkflowState(workflow.opId()).orElseThrow().opId()); @@ -241,7 +244,9 @@ public final class FilesystemPkiStoreTest { StatusObjectType.CRL, Instant.EPOCH.plusSeconds(1L), Optional.empty(), secondReference, TestObjects.emptyAttributes())); assertEquals(2, store.listStatusObjects(new PkiId("ca-status")).size()); - assertTrue(store.listPublicationRecords().isEmpty()); + try (zeroecho.pki.api.publication.PublicationCursor cursor = store.openPublicationRecords()) { + assertTrue(cursor.next(CancellationSignal.NONE).isEmpty()); + } assertFalse(Files.exists(root.resolve("staged-content").resolve(firstReference.contentId() + ".owners"))); }