From df64a4538704811a9b579a18a90326a9c6e945e9 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Wed, 9 Sep 2026 18:41:46 -0600 Subject: [PATCH 1/3] Expose pending HTTP delivery state --- .../http/HttpProxyTransportServer.java | 27 +++++++++++++++++++ .../http/HttpTransportRuntimeTest.java | 19 +++++++++++++ 2 files changed, 46 insertions(+) diff --git a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java index 2b2894ad..14545ed7 100644 --- a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java +++ b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java @@ -242,6 +242,22 @@ public boolean send(String serverId, String deliveryId, JsonEnvelope envelope) { return send(serverId, deliveryId, envelope, false); } + /** + * Returns whether this server still owns any durable proxy-to-backend + * delivery, including a publication whose durability is awaiting a same-ID + * retry. Callers can use this before retiring the HTTP transport so accepted + * work is not stranded solely because another transport was configured. + */ + public boolean hasPendingDeliveries() { + if (durableOutgoing != null) return durableOutgoing.hasPendingDeliveries(); + synchronized (backends) { + for (BackendState backend : backends.values()) { + if (backend.hasPendingOutgoing()) return true; + } + } + return false; + } + private boolean send(String serverId, String deliveryId, JsonEnvelope envelope, boolean generatedId) { if (closed || serverId == null || envelope == null) return false; try { @@ -563,6 +579,7 @@ private BackendState(String serverId, DurableOutgoingQueue durableOutgoing, private synchronized void restore(Collection deliveries) { for (HttpTransportProtocol.Delivery delivery : deliveries) outgoing.put(delivery.id(), delivery); } + private synchronized boolean hasPendingOutgoing() { return !outgoing.isEmpty(); } private boolean beginPoll(String requestedSession) { synchronized (this) { if (retired || activePoll) return false; activePoll = true; touch(); return true; } } private void endPoll() { synchronized (this) { activePoll = false; touch(); notifyAll(); } } boolean beginPollForTest() { return beginPoll("test"); } @@ -941,6 +958,16 @@ private synchronized boolean hasQuarantined(String serverId) throws IOException return quarantined != null && !quarantined.isEmpty(); } + synchronized boolean hasPendingDeliveries() { + for (Map serverFiles : files.values()) { + if (!serverFiles.isEmpty()) return true; + } + for (Map quarantined : quarantinedFiles.values()) { + if (!quarantined.isEmpty()) return true; + } + return false; + } + /** Deletes only a validated, observed-empty backend directory and makes its removal durable. */ private void deleteVerifiedEmptyDirectory(Path directory) throws IOException { if (Files.isSymbolicLink(directory) || !Files.isDirectory(directory, LinkOption.NOFOLLOW_LINKS) diff --git a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java index e2185563..f93c01bb 100644 --- a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java +++ b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java @@ -40,6 +40,23 @@ void publicConstructorsRequireDurableOutgoingDirectory() throws Exception { null, ignored -> { }, (serverId, deliveryId) -> { })); } + @Test + void pendingDeliveryStateTracksDurableQueueUntilAcknowledgement() throws Exception { + HttpTlsIdentity identity = HttpTlsIdentity.loadOrCreate(directory.resolve("pending-proxy"), "localhost"); + HttpEnrollmentAuthority authority = new HttpEnrollmentAuthority(identity, directory.resolve("pending-authority")); + try (HttpProxyTransportServer server = new HttpProxyTransportServer(new InetSocketAddress("localhost", 0), + identity, authority, directory.resolve("pending-outgoing"), ignored -> { })) { + assertFalse(server.hasPendingDeliveries()); + String deliveryId = java.util.UUID.randomUUID().toString(); + assertTrue(server.send("lobby-1", deliveryId, JsonEnvelope.builder("ordinary").build())); + assertTrue(server.hasPendingDeliveries()); + + server.backendStateForTest("lobby-1").acknowledge(java.util.List.of(deliveryId)); + + assertFalse(server.hasPendingDeliveries()); + } + } + @Test void endpointHelperSupportsIpv6Literals() throws Exception { HttpTlsIdentity identity = HttpTlsIdentity.loadOrCreate(directory.resolve("ipv6-proxy"), "::1"); @@ -275,6 +292,8 @@ void publishedOutgoingDeliveryRemainsTrackedUntilDurabilityCanBeConfirmed() thro JsonEnvelope.builder("durable").build()); assertFalse(state.enqueue(delivery), "post-publication failure must not confirm durable acceptance"); + assertTrue(queue.hasPendingDeliveries(), + "an ambiguous published file must keep the HTTP transport retained"); assertTrue(state.await("lobby-1", java.util.UUID.randomUUID().toString(), 0).messages().isEmpty(), "an uncertain publication must remain hidden until its durability retry succeeds"); assertEquals(1L, countRegularFiles(queueRoot)); From ae521f7921d91de799a40ab48d6b650ded4ef7fc Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Wed, 9 Sep 2026 19:02:23 -0600 Subject: [PATCH 2/3] Expose persisted HTTP delivery state --- .../http/HttpProxyTransportServer.java | 27 +++++++++++++++++++ .../http/HttpTransportRuntimeTest.java | 10 +++++++ 2 files changed, 37 insertions(+) diff --git a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java index 14545ed7..9aad4ff1 100644 --- a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java +++ b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java @@ -258,6 +258,33 @@ public boolean hasPendingDeliveries() { return false; } + /** + * Returns whether a stopped server's durable outgoing directory contains any + * delivery state. This is intentionally conservative: an unexpected entry is + * reported as pending so callers do not switch transports and strand data + * before the normal queue loader can validate or recover it. + */ + public static boolean hasPersistedDeliveries(Path outgoingDirectory) throws IOException { + if (outgoingDirectory == null) throw new IllegalArgumentException("HTTP outgoing queue directory is required"); + Path root = outgoingDirectory.toAbsolutePath().normalize(); + if (!Files.exists(root, LinkOption.NOFOLLOW_LINKS)) return false; + if (Files.isSymbolicLink(root) || !Files.isDirectory(root, LinkOption.NOFOLLOW_LINKS)) + throw new IOException("HTTP outgoing queue directory is invalid"); + int backendCount = 0; + try (DirectoryStream backends = Files.newDirectoryStream(root)) { + for (Path backend : backends) { + if (Files.isSymbolicLink(backend) || !Files.isDirectory(backend, LinkOption.NOFOLLOW_LINKS)) + throw new IOException("HTTP outgoing queue contains an invalid entry"); + if (++backendCount > MAX_BACKENDS) + throw new IOException("HTTP outgoing queue exceeds its backend bound"); + try (DirectoryStream entries = Files.newDirectoryStream(backend)) { + if (entries.iterator().hasNext()) return true; + } + } + } + return false; + } + private boolean send(String serverId, String deliveryId, JsonEnvelope envelope, boolean generatedId) { if (closed || serverId == null || envelope == null) return false; try { diff --git a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java index f93c01bb..6921d40f 100644 --- a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java +++ b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java @@ -57,6 +57,16 @@ void pendingDeliveryStateTracksDurableQueueUntilAcknowledgement() throws Excepti } } + @Test + void persistedDeliveryStateCanBeCheckedBeforeServerStartup() throws Exception { + Path outgoing = directory.resolve("stopped-outgoing"); + assertFalse(HttpProxyTransportServer.hasPersistedDeliveries(outgoing)); + Files.createDirectories(outgoing.resolve("lobby-1")); + assertFalse(HttpProxyTransportServer.hasPersistedDeliveries(outgoing)); + Files.writeString(outgoing.resolve("lobby-1").resolve(".pending-delivery.json"), "pending"); + assertTrue(HttpProxyTransportServer.hasPersistedDeliveries(outgoing)); + } + @Test void endpointHelperSupportsIpv6Literals() throws Exception { HttpTlsIdentity identity = HttpTlsIdentity.loadOrCreate(directory.resolve("ipv6-proxy"), "::1"); From d7584d5f8a31e3c9a385884595187510d844dbdc Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:42:59 -0600 Subject: [PATCH 3/3] Treat unreadable HTTP queues as pending --- .../servercomm/http/HttpProxyTransportServer.java | 10 ++++++++-- .../servercomm/http/HttpTransportRuntimeTest.java | 9 +++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java index 9aad4ff1..ce319c54 100644 --- a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java +++ b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java @@ -267,8 +267,14 @@ public boolean hasPendingDeliveries() { public static boolean hasPersistedDeliveries(Path outgoingDirectory) throws IOException { if (outgoingDirectory == null) throw new IllegalArgumentException("HTTP outgoing queue directory is required"); Path root = outgoingDirectory.toAbsolutePath().normalize(); - if (!Files.exists(root, LinkOption.NOFOLLOW_LINKS)) return false; - if (Files.isSymbolicLink(root) || !Files.isDirectory(root, LinkOption.NOFOLLOW_LINKS)) + java.nio.file.attribute.BasicFileAttributes rootAttributes; + try { + rootAttributes = Files.readAttributes(root, java.nio.file.attribute.BasicFileAttributes.class, + LinkOption.NOFOLLOW_LINKS); + } catch (java.nio.file.NoSuchFileException absent) { + return false; + } + if (rootAttributes.isSymbolicLink() || !rootAttributes.isDirectory()) throw new IOException("HTTP outgoing queue directory is invalid"); int backendCount = 0; try (DirectoryStream backends = Files.newDirectoryStream(root)) { diff --git a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java index 6921d40f..2cbf2177 100644 --- a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java +++ b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpTransportRuntimeTest.java @@ -67,6 +67,15 @@ void persistedDeliveryStateCanBeCheckedBeforeServerStartup() throws Exception { assertTrue(HttpProxyTransportServer.hasPersistedDeliveries(outgoing)); } + @Test + void inaccessiblePersistedDeliveryLocationIsNotReportedAsAbsent() throws Exception { + Path nonDirectoryParent = directory.resolve("queue-parent-file"); + Files.writeString(nonDirectoryParent, "not a directory"); + + assertThrows(java.io.IOException.class, + () -> HttpProxyTransportServer.hasPersistedDeliveries(nonDirectoryParent.resolve("outgoing"))); + } + @Test void endpointHelperSupportsIpv6Literals() throws Exception { HttpTlsIdentity identity = HttpTlsIdentity.loadOrCreate(directory.resolve("ipv6-proxy"), "::1");