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")));
}