feat(pki): reconcile recovered signing operations
Add bounded durable reconciliation with retry metadata, fencing-safe status and cancellation handling, and server-managed background recovery. Include versioned persistence migration, bounded keyset paging, lifecycle-safe worker shutdown, redacted diagnostics, and restart/failure coverage. Closes #10
This commit is contained in:
@@ -70,6 +70,7 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
private final Optional<PkiServerAuthenticator> publicAuthenticator;
|
||||
private final Optional<PublicRepositoryTransport> publicTransport;
|
||||
private final Optional<AcmeTransport> acmeTransport;
|
||||
private final Optional<SigningReconciliationWorker> signingReconciliation;
|
||||
private final AtomicReference<State> state;
|
||||
|
||||
private PkiHttpsServer(PkiServerConfiguration configuration, ServerRealmContext realm,
|
||||
@@ -77,6 +78,7 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
Optional<PkiServerAuthenticator> publicAuthenticator,
|
||||
Optional<PublicRepositoryTransport> publicTransport,
|
||||
Optional<AcmeTransport> acmeTransport,
|
||||
Optional<SigningReconciliationWorker> signingReconciliation,
|
||||
AtomicReference<State> state) {
|
||||
this.configuration = configuration;
|
||||
this.realm = realm;
|
||||
@@ -85,6 +87,7 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
this.publicAuthenticator = publicAuthenticator;
|
||||
this.publicTransport = publicTransport;
|
||||
this.acmeTransport = acmeTransport;
|
||||
this.signingReconciliation = signingReconciliation;
|
||||
this.state = state;
|
||||
}
|
||||
|
||||
@@ -116,6 +119,7 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
PkiServerAuthenticator publicAuthenticator = null;
|
||||
PublicRepositoryTransport publicTransport = null;
|
||||
AcmeTransport acmeTransport = null;
|
||||
SigningReconciliationWorker signingReconciliation = null;
|
||||
try {
|
||||
SSLContext tls = PkiHttpsTransport.resolveTls(exact, loader);
|
||||
realm = ServerRealmContext.open(exact.realm(), runtimeDependencies, clock, random);
|
||||
@@ -160,14 +164,22 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
acmeTransport = AcmeTransport.start(acmeConfiguration, sharedRealm, clock, random, loader,
|
||||
() -> state.get() == State.READY && sharedRealm.state() == ServerRealmContext.State.OPEN);
|
||||
}
|
||||
state.set(State.READY);
|
||||
if (exact.signingReconciliation().enabled()) {
|
||||
signingReconciliation = new SigningReconciliationWorker(realm, exact.signingReconciliation(),
|
||||
exact.execution().forcedShutdown(), clock);
|
||||
}
|
||||
realm.auditTransport("SERVER_READY", "system", Map.of("state", "READY"));
|
||||
return new PkiHttpsServer(exact, realm, authenticator, transport,
|
||||
PkiHttpsServer server = new PkiHttpsServer(exact, realm, authenticator, transport,
|
||||
Optional.ofNullable(publicAuthenticator), Optional.ofNullable(publicTransport),
|
||||
Optional.ofNullable(acmeTransport), state);
|
||||
Optional.ofNullable(acmeTransport), Optional.ofNullable(signingReconciliation), state);
|
||||
if (signingReconciliation != null) {
|
||||
signingReconciliation.start();
|
||||
}
|
||||
state.set(State.READY);
|
||||
return server;
|
||||
} catch (RuntimeException | Error primary) {
|
||||
closePartial(acmeTransport, publicTransport, publicAuthenticator, transport, realm, authenticator,
|
||||
state, primary);
|
||||
closePartial(signingReconciliation, acmeTransport, publicTransport, publicAuthenticator, transport, realm,
|
||||
authenticator, state, primary);
|
||||
throw primary;
|
||||
}
|
||||
}
|
||||
@@ -221,8 +233,10 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
*/
|
||||
@Override
|
||||
public void close() throws Exception {
|
||||
if (!state.compareAndSet(State.READY, State.QUIESCING)
|
||||
State observed = state.get();
|
||||
if (observed != State.QUIESCING && !state.compareAndSet(State.READY, State.QUIESCING)
|
||||
&& !state.compareAndSet(State.STARTING, State.QUIESCING)) return;
|
||||
signingReconciliation.ifPresent(SigningReconciliationWorker::stopScheduling);
|
||||
Throwable primary = null;
|
||||
try {
|
||||
realm.auditTransport("SERVER_SHUTDOWN", "system", Map.of("state", "QUIESCING"));
|
||||
@@ -253,6 +267,15 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
} catch (Throwable failure) {
|
||||
primary = suppress(primary, failure);
|
||||
}
|
||||
try {
|
||||
if (signingReconciliation.isPresent()) {
|
||||
signingReconciliation.orElseThrow().awaitStopped();
|
||||
}
|
||||
} catch (Throwable failure) {
|
||||
if (primary != null && primary != failure) failure.addSuppressed(primary);
|
||||
rethrow(failure);
|
||||
return;
|
||||
}
|
||||
if (publicAuthenticator.isPresent()) primary = close(publicAuthenticator.orElseThrow(), primary);
|
||||
primary = close(authenticator, primary);
|
||||
primary = close(realm, primary);
|
||||
@@ -260,9 +283,21 @@ public final class PkiHttpsServer implements AutoCloseable {
|
||||
rethrow(primary);
|
||||
}
|
||||
|
||||
private static void closePartial(AcmeTransport acmeTransport, PublicRepositoryTransport publicTransport,
|
||||
private static void closePartial(SigningReconciliationWorker signingReconciliation, AcmeTransport acmeTransport,
|
||||
PublicRepositoryTransport publicTransport,
|
||||
PkiServerAuthenticator publicAuthenticator, PkiHttpsTransport transport, ServerRealmContext realm,
|
||||
PkiServerAuthenticator authenticator, AtomicReference<State> state, Throwable primary) {
|
||||
if (signingReconciliation != null) {
|
||||
try {
|
||||
signingReconciliation.close();
|
||||
} catch (Throwable failure) {
|
||||
if (primary != failure) primary.addSuppressed(failure);
|
||||
close(acmeTransport, primary);
|
||||
close(publicTransport, primary);
|
||||
close(transport, primary);
|
||||
return;
|
||||
}
|
||||
}
|
||||
primary = close(acmeTransport, primary);
|
||||
primary = close(publicTransport, primary);
|
||||
primary = close(publicAuthenticator, primary);
|
||||
|
||||
@@ -61,14 +61,16 @@ import zeroecho.pki.spi.ProviderConfig;
|
||||
* @param runtime process-local capability references
|
||||
* @param publicListener optional separately bounded public repository listener
|
||||
* @param acmeListener optional separately bounded ACME protocol listener
|
||||
* @param signingReconciliation bounded recovered-signing owner policy
|
||||
*/
|
||||
@SuppressWarnings("PMD")
|
||||
public record PkiServerConfiguration(int version, String serverName, ServerRealmConfiguration realm,
|
||||
Listener listener, Authentication authentication, Execution execution, RuntimeCapabilities runtime,
|
||||
Optional<PublicListener> publicListener, Optional<AcmeListener> acmeListener) {
|
||||
Optional<PublicListener> publicListener, Optional<AcmeListener> acmeListener,
|
||||
SigningReconciliation signingReconciliation) {
|
||||
|
||||
/** Current server configuration schema. */
|
||||
public static final int CURRENT_VERSION = 5;
|
||||
public static final int CURRENT_VERSION = 6;
|
||||
|
||||
/** Validates all security-sensitive fields before resource allocation. */
|
||||
public PkiServerConfiguration {
|
||||
@@ -81,6 +83,7 @@ public record PkiServerConfiguration(int version, String serverName, ServerRealm
|
||||
Objects.requireNonNull(runtime, "runtime");
|
||||
publicListener = Objects.requireNonNull(publicListener, "publicListener");
|
||||
acmeListener = Objects.requireNonNull(acmeListener, "acmeListener");
|
||||
Objects.requireNonNull(signingReconciliation, "signingReconciliation");
|
||||
if (!listener.clientCertificateRequired()) {
|
||||
throw new IllegalArgumentException("Administrative HTTPS requires client certificates");
|
||||
}
|
||||
@@ -110,14 +113,53 @@ public record PkiServerConfiguration(int version, String serverName, ServerRealm
|
||||
Listener listener, Authentication authentication, Execution execution, RuntimeCapabilities runtime,
|
||||
Optional<PublicListener> publicListener) {
|
||||
this(version, serverName, realm, listener, authentication, execution, runtime, publicListener,
|
||||
Optional.empty());
|
||||
Optional.empty(), SigningReconciliation.defaults());
|
||||
}
|
||||
|
||||
/** Creates a configuration with default signing reconciliation. */
|
||||
public PkiServerConfiguration(int version, String serverName, ServerRealmConfiguration realm,
|
||||
Listener listener, Authentication authentication, Execution execution, RuntimeCapabilities runtime,
|
||||
Optional<PublicListener> publicListener, Optional<AcmeListener> acmeListener) {
|
||||
this(version, serverName, realm, listener, authentication, execution, runtime, publicListener, acmeListener,
|
||||
SigningReconciliation.defaults());
|
||||
}
|
||||
|
||||
/** Creates a configuration with the public repository listener disabled. */
|
||||
public PkiServerConfiguration(int version, String serverName, ServerRealmConfiguration realm,
|
||||
Listener listener, Authentication authentication, Execution execution, RuntimeCapabilities runtime) {
|
||||
this(version, serverName, realm, listener, authentication, execution, runtime, Optional.empty(),
|
||||
Optional.empty());
|
||||
Optional.empty(), SigningReconciliation.defaults());
|
||||
}
|
||||
|
||||
/**
|
||||
* Bounded recovered-signing reconciliation policy.
|
||||
*
|
||||
* @param enabled whether the realm owns a reconciliation scheduler
|
||||
* @param maximumRecords maximum records examined per pass
|
||||
* @param maximumProviderCalls maximum provider calls per pass
|
||||
* @param passTimeout positive complete-pass timeout
|
||||
* @param providerCallTimeout positive per-call timeout no greater than the pass timeout
|
||||
* @param idleInterval positive delay after each completed pass
|
||||
*/
|
||||
public record SigningReconciliation(boolean enabled, int maximumRecords, int maximumProviderCalls,
|
||||
Duration passTimeout, Duration providerCallTimeout, Duration idleInterval) {
|
||||
/** Validates all reconciliation bounds. */
|
||||
public SigningReconciliation {
|
||||
bounded(maximumRecords, 1, 4096, "signing reconciliation record count");
|
||||
bounded(maximumProviderCalls, 1, maximumRecords, "signing reconciliation provider call count");
|
||||
positive(passTimeout, Duration.ofMinutes(5), "signing reconciliation pass timeout");
|
||||
positive(providerCallTimeout, Duration.ofMinutes(5), "signing reconciliation provider call timeout");
|
||||
positive(idleInterval, Duration.ofMinutes(5), "signing reconciliation idle interval");
|
||||
if (providerCallTimeout.compareTo(passTimeout) > 0) {
|
||||
throw new IllegalArgumentException("Signing reconciliation provider timeout exceeds pass timeout");
|
||||
}
|
||||
}
|
||||
|
||||
/** @return version-6 production defaults */
|
||||
public static SigningReconciliation defaults() {
|
||||
return new SigningReconciliation(true, 256, 64, Duration.ofSeconds(30), Duration.ofSeconds(10),
|
||||
Duration.ofSeconds(2));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -86,12 +86,19 @@ public final class PkiServerConfigurationCodec {
|
||||
public static PkiServerConfiguration decode(byte[] document) {
|
||||
Fields root = Fields.of(StrictJson.parse(document, MAXIMUM_CONFIGURATION_BYTES));
|
||||
root.allowed(Set.of("version", "serverName", "realm", "listener", "authentication", "execution",
|
||||
"runtime", "publicListener", "acmeListener"));
|
||||
PkiServerConfiguration configuration = new PkiServerConfiguration(root.integer("version"),
|
||||
"runtime", "publicListener", "acmeListener", "signingReconciliation"));
|
||||
int encodedVersion = root.integer("version");
|
||||
if (encodedVersion != 5 && encodedVersion != PkiServerConfiguration.CURRENT_VERSION) {
|
||||
throw new IllegalArgumentException("Unsupported server schema version");
|
||||
}
|
||||
PkiServerConfiguration.SigningReconciliation reconciliation = encodedVersion == 5
|
||||
? PkiServerConfiguration.SigningReconciliation.defaults()
|
||||
: signingReconciliation(root.object("signingReconciliation"));
|
||||
PkiServerConfiguration configuration = new PkiServerConfiguration(PkiServerConfiguration.CURRENT_VERSION,
|
||||
root.text("serverName"), realm(root.object("realm")), listener(root.object("listener")),
|
||||
authentication(root.object("authentication")), execution(root.object("execution")),
|
||||
runtime(root.object("runtime")), publicListener(root.object("publicListener")),
|
||||
acmeListener(root.object("acmeListener")));
|
||||
acmeListener(root.object("acmeListener")), reconciliation);
|
||||
root.complete();
|
||||
return configuration;
|
||||
}
|
||||
@@ -273,6 +280,18 @@ public final class PkiServerConfigurationCodec {
|
||||
return result;
|
||||
}
|
||||
|
||||
private static PkiServerConfiguration.SigningReconciliation signingReconciliation(Fields value) {
|
||||
value.exact("enabled", "maxRecords", "maxCalls", "passTimeoutMillis", "providerCallTimeoutMillis",
|
||||
"idleIntervalMillis");
|
||||
PkiServerConfiguration.SigningReconciliation result = new PkiServerConfiguration.SigningReconciliation(
|
||||
value.bool("enabled"), value.integer("maxRecords"), value.integer("maxCalls"),
|
||||
Duration.ofMillis(value.longValue("passTimeoutMillis")),
|
||||
Duration.ofMillis(value.longValue("providerCallTimeoutMillis")),
|
||||
Duration.ofMillis(value.longValue("idleIntervalMillis")));
|
||||
value.complete();
|
||||
return result;
|
||||
}
|
||||
|
||||
private static Optional<PkiServerConfiguration.PublicListener> publicListener(Fields value) {
|
||||
value.allowed(Set.of("enabled", "address", "port", "tlsProvider", "allowPlaintextLoopback", "authentication",
|
||||
"maximumHeaderBytes", "maximumBodyBytes", "execution", "maximumStreamDurationMillis",
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
/*******************************************************************************
|
||||
* Copyright (C) 2026, Leo Galambos
|
||||
* All rights reserved.
|
||||
******************************************************************************/
|
||||
package zeroecho.pki.server;
|
||||
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import zeroecho.pki.application.SigningReconciliationRequest;
|
||||
import zeroecho.pki.application.SigningReconciliationResult;
|
||||
import zeroecho.pki.application.PkiSession;
|
||||
|
||||
/** Single-threaded, self-rescheduling signing recovery owner for one realm. */
|
||||
@SuppressWarnings({ "PMD.DoNotUseThreads", "PMD.AvoidUsingVolatile", "PMD.CommentDefaultAccessModifier" })
|
||||
final class SigningReconciliationWorker implements AutoCloseable {
|
||||
private final PkiSession session;
|
||||
private final AuditTransport audit;
|
||||
private final PkiServerConfiguration.SigningReconciliation configuration;
|
||||
private final Duration forcedShutdown;
|
||||
private final Clock clock;
|
||||
private final ScheduledExecutorService executor;
|
||||
private final AtomicBoolean cancelled = new AtomicBoolean();
|
||||
private Optional<String> cursor = Optional.empty();
|
||||
private volatile Future<?> scheduled;
|
||||
|
||||
/* default */ SigningReconciliationWorker(ServerRealmContext realm,
|
||||
PkiServerConfiguration.SigningReconciliation configuration, Duration forcedShutdown, Clock clock) {
|
||||
this(Objects.requireNonNull(realm, "realm").session(), realm::auditTransport, configuration, forcedShutdown,
|
||||
clock);
|
||||
}
|
||||
|
||||
/* default */ SigningReconciliationWorker(PkiSession session, AuditTransport audit,
|
||||
PkiServerConfiguration.SigningReconciliation configuration, Duration forcedShutdown, Clock clock) {
|
||||
this.session = Objects.requireNonNull(session, "session");
|
||||
this.audit = Objects.requireNonNull(audit, "audit");
|
||||
this.configuration = Objects.requireNonNull(configuration, "configuration");
|
||||
this.forcedShutdown = Objects.requireNonNull(forcedShutdown, "forcedShutdown");
|
||||
this.clock = Objects.requireNonNull(clock, "clock");
|
||||
this.executor = Executors.newSingleThreadScheduledExecutor(runnable -> {
|
||||
Thread thread = new Thread(runnable, "zeroecho-signing-reconciliation");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
}
|
||||
|
||||
/* default */ void start() {
|
||||
if (cancelled.get() || scheduled != null) {
|
||||
throw new IllegalStateException("Signing reconciliation worker cannot start");
|
||||
}
|
||||
scheduled = executor.submit(this::runPass);
|
||||
}
|
||||
|
||||
@SuppressWarnings("PMD.AvoidCatchingGenericException")
|
||||
private void runPass() {
|
||||
if (cancelled.get()) {
|
||||
return;
|
||||
}
|
||||
Instant deadline = clock.instant().plus(configuration.passTimeout());
|
||||
Map<String, String> auditDetails;
|
||||
try {
|
||||
SigningReconciliationResult result = session.reconcileSigning(new SigningReconciliationRequest(
|
||||
cursor, configuration.maximumRecords(), configuration.maximumProviderCalls(), deadline,
|
||||
configuration.providerCallTimeout(), cancelled::get));
|
||||
cursor = result.nextCursor();
|
||||
auditDetails = Map.of(
|
||||
"code", result.retryable() == 0 ? "SIGNING_RECONCILIATION_OK" : "SIGNING_RECONCILIATION_RETRY",
|
||||
"examined", Integer.toString(result.examined()),
|
||||
"progressed", Integer.toString(result.progressed()),
|
||||
"unresolved", Integer.toString(result.unresolved()),
|
||||
"retryable", Integer.toString(result.retryable()),
|
||||
"endReached", Boolean.toString(result.endReached()));
|
||||
} catch (RuntimeException failure) {
|
||||
auditDetails = Map.of("code", "SIGNING_RECONCILIATION_PASS_FAILED");
|
||||
}
|
||||
try {
|
||||
audit.record("SIGNING_RECONCILIATION_PASS", "system", auditDetails);
|
||||
} catch (RuntimeException ignored) {
|
||||
// The pass remains bounded and future passes remain authoritative.
|
||||
} finally {
|
||||
if (!cancelled.get()) {
|
||||
scheduled = executor.schedule(this::runPass, configuration.idleInterval().toNanos(),
|
||||
TimeUnit.NANOSECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws InterruptedException {
|
||||
stopScheduling();
|
||||
awaitStopped();
|
||||
}
|
||||
|
||||
/* default */ void stopScheduling() {
|
||||
if (cancelled.compareAndSet(false, true)) {
|
||||
Future<?> current = scheduled;
|
||||
if (current != null) {
|
||||
current.cancel(true);
|
||||
}
|
||||
executor.shutdownNow();
|
||||
}
|
||||
}
|
||||
|
||||
/* default */ void awaitStopped() throws InterruptedException {
|
||||
if (!cancelled.get()) {
|
||||
throw new IllegalStateException("Signing reconciliation worker is still accepting work");
|
||||
}
|
||||
if (!executor.awaitTermination(forcedShutdown.toNanos(), TimeUnit.NANOSECONDS)) {
|
||||
throw new IllegalStateException(
|
||||
"Signing reconciliation shutdown timed out: code=SIGNING_RECONCILIATION_SHUTDOWN_TIMEOUT");
|
||||
}
|
||||
}
|
||||
|
||||
/** Safe aggregate reconciliation audit boundary. */
|
||||
@FunctionalInterface
|
||||
/* default */
|
||||
interface AuditTransport {
|
||||
/** Records one aggregate event without record or provider details. */
|
||||
void record(String eventType, String principal, Map<String, String> details);
|
||||
}
|
||||
}
|
||||
@@ -755,7 +755,10 @@ class AcmeEndToEndTest {
|
||||
+ "\"port\":" + configuration.listener().port() + ",\"tlsProvider\":" + tlsJson
|
||||
+ ",\"clientCertificateRequired\":true,\"maximumHeaderBytes\":16384,"
|
||||
+ "\"maximumBodyBytes\":1048576},\"authentication\":" + authenticationJson
|
||||
+ ",\"execution\":" + executionJson() + ",\"publicListener\":{\"enabled\":true,"
|
||||
+ ",\"execution\":" + executionJson()
|
||||
+ ",\"signingReconciliation\":{\"enabled\":true,\"maxRecords\":256,\"maxCalls\":64,"
|
||||
+ "\"passTimeoutMillis\":30000,\"providerCallTimeoutMillis\":10000,"
|
||||
+ "\"idleIntervalMillis\":2000},\"publicListener\":{\"enabled\":true,"
|
||||
+ "\"address\":\"127.0.0.1\",\"port\":" + publicListener.port()
|
||||
+ ",\"tlsProvider\":" + tlsJson + ",\"allowPlaintextLoopback\":false,"
|
||||
+ "\"authentication\":" + authenticationJson
|
||||
|
||||
@@ -121,8 +121,24 @@ class PkiHttpsServerTest {
|
||||
valid.realm(), valid.listener(), valid.authentication(), valid.execution(), valid.runtime()));
|
||||
String production = java.nio.file.Files.readString(java.nio.file.Path.of(
|
||||
"..", "docs", "pki-server-production-example.json"));
|
||||
assertTrue(PkiServerConfigurationCodec.decode(production.getBytes(
|
||||
java.nio.charset.StandardCharsets.UTF_8)).publicListener().isEmpty());
|
||||
PkiServerConfiguration decoded = PkiServerConfigurationCodec.decode(production.getBytes(
|
||||
java.nio.charset.StandardCharsets.UTF_8));
|
||||
assertTrue(decoded.publicListener().isEmpty());
|
||||
assertEquals(PkiServerConfiguration.SigningReconciliation.defaults(), decoded.signingReconciliation());
|
||||
String versionFive = production.replace("\"version\": 6", "\"version\": 5")
|
||||
.replace(" \"signingReconciliation\": {\"enabled\": true, \"maxRecords\": 256, "
|
||||
+ "\"maxCalls\": 64, \"passTimeoutMillis\": 30000, "
|
||||
+ "\"providerCallTimeoutMillis\": 10000, \"idleIntervalMillis\": 2000},\n", "");
|
||||
PkiServerConfiguration migrated = PkiServerConfigurationCodec.decode(
|
||||
versionFive.getBytes(java.nio.charset.StandardCharsets.UTF_8));
|
||||
assertEquals(PkiServerConfiguration.CURRENT_VERSION, migrated.version());
|
||||
assertEquals(PkiServerConfiguration.SigningReconciliation.defaults(), migrated.signingReconciliation());
|
||||
assertThrows(IllegalArgumentException.class,
|
||||
() -> new PkiServerConfiguration.SigningReconciliation(true, 4, 5, Duration.ofSeconds(30),
|
||||
Duration.ofSeconds(10), Duration.ofSeconds(2)));
|
||||
assertThrows(IllegalArgumentException.class,
|
||||
() -> new PkiServerConfiguration.SigningReconciliation(true, 4, 4, Duration.ofSeconds(5),
|
||||
Duration.ofSeconds(10), Duration.ofSeconds(2)));
|
||||
assertThrows(IllegalArgumentException.class, () -> PkiServerConfigurationCodec.decode(production
|
||||
.replace("\"publicListener\": {\"enabled\": false},\n", "")
|
||||
.getBytes(java.nio.charset.StandardCharsets.UTF_8)));
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
/*******************************************************************************
|
||||
* Copyright (C) 2026, Leo Galambos
|
||||
* All rights reserved.
|
||||
******************************************************************************/
|
||||
package zeroecho.pki.server;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneOffset;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import zeroecho.pki.application.PkiSession;
|
||||
import zeroecho.pki.application.SigningReconciliationResult;
|
||||
|
||||
final class SigningReconciliationWorkerTest {
|
||||
|
||||
@Test
|
||||
void timeoutStopsSchedulingAndNeverClosesSessionUnderViolatingCall() throws Exception {
|
||||
CountDownLatch entered = new CountDownLatch(1);
|
||||
AtomicBoolean release = new AtomicBoolean();
|
||||
AtomicInteger calls = new AtomicInteger();
|
||||
AtomicInteger closes = new AtomicInteger();
|
||||
PkiSession session = session((method, arguments) -> {
|
||||
if ("reconcileSigning".equals(method)) {
|
||||
calls.incrementAndGet();
|
||||
entered.countDown();
|
||||
while (!release.get()) {
|
||||
try {
|
||||
Thread.sleep(1L);
|
||||
} catch (InterruptedException ignored) {
|
||||
// Deliberately violates cooperative cancellation for this lifecycle test.
|
||||
}
|
||||
}
|
||||
return emptyResult();
|
||||
}
|
||||
if ("close".equals(method)) {
|
||||
closes.incrementAndGet();
|
||||
}
|
||||
return null;
|
||||
});
|
||||
SigningReconciliationWorker worker = new SigningReconciliationWorker(session, (type, principal, details) -> {
|
||||
// Aggregate audit is irrelevant to the blocking lifecycle assertion.
|
||||
}, configuration(Duration.ofSeconds(1)), Duration.ofMillis(20), fixedClock());
|
||||
worker.start();
|
||||
assertTrue(entered.await(5, TimeUnit.SECONDS));
|
||||
|
||||
worker.stopScheduling();
|
||||
IllegalStateException timeout = assertThrows(IllegalStateException.class, worker::awaitStopped);
|
||||
assertTrue(timeout.getMessage().contains("SIGNING_RECONCILIATION_SHUTDOWN_TIMEOUT"));
|
||||
assertEquals(1, calls.get());
|
||||
assertEquals(0, closes.get());
|
||||
|
||||
release.set(true);
|
||||
worker.awaitStopped();
|
||||
assertEquals(1, calls.get());
|
||||
assertEquals(0, closes.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void subMillisecondIdleIntervalSelfSchedulesWithoutOverlap() throws Exception {
|
||||
AtomicInteger active = new AtomicInteger();
|
||||
AtomicInteger maximumActive = new AtomicInteger();
|
||||
CountDownLatch twoPasses = new CountDownLatch(2);
|
||||
PkiSession session = session((method, arguments) -> {
|
||||
if ("reconcileSigning".equals(method)) {
|
||||
int current = active.incrementAndGet();
|
||||
maximumActive.accumulateAndGet(current, Math::max);
|
||||
twoPasses.countDown();
|
||||
active.decrementAndGet();
|
||||
return emptyResult();
|
||||
}
|
||||
return null;
|
||||
});
|
||||
SigningReconciliationWorker worker = new SigningReconciliationWorker(session, (type, principal, details) -> {
|
||||
// Safe aggregate event accepted.
|
||||
}, configuration(Duration.ofNanos(1)), Duration.ofSeconds(1), fixedClock());
|
||||
worker.start();
|
||||
assertTrue(twoPasses.await(5, TimeUnit.SECONDS));
|
||||
worker.close();
|
||||
assertEquals(1, maximumActive.get());
|
||||
}
|
||||
|
||||
private static PkiServerConfiguration.SigningReconciliation configuration(Duration idle) {
|
||||
return new PkiServerConfiguration.SigningReconciliation(true, 1, 1, Duration.ofSeconds(1),
|
||||
Duration.ofMillis(100), idle);
|
||||
}
|
||||
|
||||
private static SigningReconciliationResult emptyResult() {
|
||||
return new SigningReconciliationResult(Optional.empty(), 0, 0, 0, 0, true);
|
||||
}
|
||||
|
||||
private static Clock fixedClock() {
|
||||
return Clock.fixed(Instant.parse("2026-08-05T00:00:00Z"), ZoneOffset.UTC);
|
||||
}
|
||||
|
||||
private static PkiSession session(SessionInvocation invocation) {
|
||||
return (PkiSession) Proxy.newProxyInstance(PkiSession.class.getClassLoader(), new Class<?>[] { PkiSession.class },
|
||||
(proxy, method, arguments) -> {
|
||||
if (method.getDeclaringClass() == Object.class) {
|
||||
return switch (method.getName()) {
|
||||
case "toString" -> "test-pki-session";
|
||||
case "hashCode" -> System.identityHashCode(proxy);
|
||||
case "equals" -> proxy == arguments[0];
|
||||
default -> null;
|
||||
};
|
||||
}
|
||||
return invocation.invoke(method.getName(), arguments);
|
||||
});
|
||||
}
|
||||
|
||||
@FunctionalInterface
|
||||
private interface SessionInvocation {
|
||||
Object invoke(String method, Object[] arguments) throws Throwable;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user