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 2b2894a..ce319c5 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,55 @@ 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; + } + + /** + * 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(); + 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)) { + 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 { @@ -563,6 +612,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 +991,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 e218556..2cbf217 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,42 @@ 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 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 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"); @@ -275,6 +311,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));