From a4c409dbac6a1997de774fd99d1d381331ce0e8f Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 15:36:45 -0600 Subject: [PATCH 01/33] Bound connection throttle state --- .../vexsoftware/votifier/net/VoteConnectionHandler.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index f602927..a57980d 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -57,6 +57,14 @@ public Vote handle(Socket socket) { address = accepted.getRemoteSocketAddress() == null ? "/" + remoteIp : accepted.getRemoteSocketAddress().toString(); + throttleKey = "tunnel:" + remoteIp; + tunnelMode = throttleService.isTunnelMode(remoteIp); + if (throttleService.isBlocked(throttleKey)) { + throttleService.logWarning(receiver, "throttle|" + throttleKey, + "Votifier rejected a throttled connection from " + remoteIp); + return null; + } + receiver.debug("Accepted connection from: " + address); accepted.setSoTimeout(5000); @@ -73,7 +81,6 @@ public Vote handle(Socket socket) { realIp = proxyResult.getRealIp(); realIpKnown = realIp != null && !realIp.isEmpty(); - tunnelMode = throttleService.isTunnelMode(remoteIp); throttleKey = realIpKnown ? "ip:" + realIp : "tunnel:" + remoteIp; if (throttleService.isBlocked(throttleKey)) { From 36d178cadb3cb563d76dcf4d345b38f1bafdf982 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 15:36:56 -0600 Subject: [PATCH 02/33] Bound connection throttle state --- .../votifier/net/VoteThrottleService.java | 41 ++++++++++++++++++- 1 file changed, 40 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 17e6a18..ac473c2 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -10,6 +10,7 @@ import java.util.concurrent.ConcurrentHashMap; public class VoteThrottleService { + private static final int MAX_TRACKED_KEYS = 4096; private static final class LogState { private volatile long lastLogMs; @@ -103,6 +104,7 @@ public String allowLog(String key, String msg) { long now = System.currentTimeMillis(); long windowMs = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; + trimLogStates(now); LogState state = logStates.get(key); if (state == null) { LogState created = new LogState(); @@ -139,6 +141,7 @@ public void logGenericError(String remoteIp, Exception ex) { } private ThrottleState getThrottleState(String key) { + trimThrottleStates(System.currentTimeMillis()); ThrottleState state = throttleStates.get(key); if (state == null) { ThrottleState created = new ThrottleState(); @@ -148,4 +151,40 @@ private ThrottleState getThrottleState(String key) { } return state; } -} \ No newline at end of file + + private void trimLogStates(long now) { + if (logStates.size() < MAX_TRACKED_KEYS) { + return; + } + long expiry = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; + for (java.util.Map.Entry entry : logStates.entrySet()) { + if (now - entry.getValue().lastLogMs >= expiry) { + logStates.remove(entry.getKey(), entry.getValue()); + } + } + removeOneIfFull(logStates); + } + + private void trimThrottleStates(long now) { + if (throttleStates.size() < MAX_TRACKED_KEYS) { + return; + } + for (java.util.Map.Entry entry : throttleStates.entrySet()) { + ThrottleState state = entry.getValue(); + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && now - state.windowStartMs > config.windowMs) { + throttleStates.remove(entry.getKey(), state); + } + } + removeOneIfFull(throttleStates); + } + + private static void removeOneIfFull(ConcurrentHashMap states) { + if (states.size() >= MAX_TRACKED_KEYS) { + java.util.Iterator iterator = states.keySet().iterator(); + if (iterator.hasNext()) { + states.remove(iterator.next()); + } + } + } +} From 2c9ffa20ac2e9cc847b918dd591e7da402d4177d Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 15:37:08 -0600 Subject: [PATCH 03/33] Bound connection throttle state --- .../votifierplus/tests/VoteConnectionHandlerTest.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java index 2232323..c73070e 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java @@ -260,7 +260,7 @@ public Vote call() { } @Test - public void testHandleBlockedConnectionReturnsNull() throws Exception { + public void testBlockedConnectionIsRejectedBeforeHandshake() throws Exception { receiver.setUseTokens(false); ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "10s", 1, "30s", 1, "30s", false, 999, "1s", "60s"); @@ -285,7 +285,7 @@ public Vote call() { new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); String handshake = clientReader.readLine(); - assertEquals("VOTIFIER 1", handshake); + assertNull(handshake, "Blocked peers must not consume handshake or payload resources"); Vote vote = future.get(); assertNull(vote); @@ -493,4 +493,4 @@ public Vote call() { assertTrue(!clientReader.ready(), "Did not expect an OK response for TestVote"); } } -} \ No newline at end of file +} From c614557fbc6c7627867437b72216f2f7a28ccde9 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 15:37:22 -0600 Subject: [PATCH 04/33] Bound connection throttle state --- .../tests/VoteReceiverThrottleTest.java | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index ac14fee..ec357d6 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -7,6 +7,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.Collections; + import java.lang.reflect.Field; + import java.util.Map; import org.junit.jupiter.api.Test; @@ -165,4 +167,22 @@ public void testTunnelModeDetection() { assertTrue(service.isTunnelMode("10.0.0.1")); assertFalse(service.isTunnelMode("10.0.0.2")); } -} \ No newline at end of file + + @Test + public void testAttackerControlledKeysAreBounded() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 5000; i++) { + service.fail("ip:" + i, false, true); + service.allowLog("log:" + i, "message"); + } + assertTrue(mapSize(service, "throttleStates") <= 4096); + assertTrue(mapSize(service, "logStates") <= 4096); + } + + private static int mapSize(VoteThrottleService service, String fieldName) throws Exception { + Field field = VoteThrottleService.class.getDeclaredField(fieldName); + field.setAccessible(true); + return ((Map) field.get(service)).size(); + } +} From 43bdd30d83019dd7e3ba1743e93a7e274be6653e Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 16:37:04 -0600 Subject: [PATCH 05/33] Preserve shared tunnel client isolation --- .../com/vexsoftware/votifier/net/VoteConnectionHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index a57980d..d95e2ca 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -59,7 +59,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; tunnelMode = throttleService.isTunnelMode(remoteIp); - if (throttleService.isBlocked(throttleKey)) { + if (!tunnelMode && throttleService.isBlocked(throttleKey)) { throttleService.logWarning(receiver, "throttle|" + throttleKey, "Votifier rejected a throttled connection from " + remoteIp); return null; From 630374d2742996a0bb5603c0dd196f7e0eaf446f Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 16:37:18 -0600 Subject: [PATCH 06/33] Preserve shared tunnel client isolation --- .../tests/VoteConnectionHandlerTest.java | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java index c73070e..c1916d9 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java @@ -292,6 +292,27 @@ public Vote call() { } } + @Test + public void testConfiguredTunnelCanProvideClientIdentityBeforeThrottleDecision() throws Exception { + receiver.setUseTokens(false); + ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("127.0.0.1"), "10s", 1, "30s", 1, + "30s", false, 999, "1s", "60s"); + VoteThrottleService throttleService = new VoteThrottleService(config); + VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); + throttleService.fail("tunnel:127.0.0.1", true, false); + + try (ServerSocket serverSocket = new ServerSocket(0); + Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); + Socket accepted = serverSocket.accept()) { + Future future = executor.submit(() -> handler.handle(accepted)); + BufferedReader reader = new BufferedReader( + new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); + assertEquals("VOTIFIER 1", reader.readLine(), "Shared tunnels must reach proxy-header detection"); + client.close(); + assertNull(future.get()); + } + } + @Test public void testHandlePresentV1PayloadSkipsHandshake() throws Exception { receiver.setUseTokens(false); From bf1d6862fe224748ece6dc94cb000aebd0a2f913 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 16:53:42 -0600 Subject: [PATCH 07/33] Preserve tracked throttle state at capacity --- .../java/com/vexsoftware/votifier/net/VoteThrottleService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index ac473c2..7796a20 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -141,9 +141,9 @@ public void logGenericError(String remoteIp, Exception ex) { } private ThrottleState getThrottleState(String key) { - trimThrottleStates(System.currentTimeMillis()); ThrottleState state = throttleStates.get(key); if (state == null) { + trimThrottleStates(System.currentTimeMillis()); ThrottleState created = new ThrottleState(); created.windowStartMs = System.currentTimeMillis(); ThrottleState existing = throttleStates.putIfAbsent(key, created); From 5709c9bb3d5d60db74fa0b6465a957d8e3800620 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 16:53:53 -0600 Subject: [PATCH 08/33] Preserve tracked throttle state at capacity --- .../tests/VoteReceiverThrottleTest.java | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index ec357d6..e34cb3f 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -180,9 +180,27 @@ public void testAttackerControlledKeysAreBounded() throws Exception { assertTrue(mapSize(service, "logStates") <= 4096); } + @Test + public void testExistingKeyDoesNotEvictAnotherStateAtCapacity() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + Map states = stateMap(service, "throttleStates"); + Object existing = states.get("ip:100"); + service.fail("ip:100", false, false); + assertTrue(existing == states.get("ip:100")); + assertEquals(4096, states.size()); + } + private static int mapSize(VoteThrottleService service, String fieldName) throws Exception { + return stateMap(service, fieldName).size(); + } + + private static Map stateMap(VoteThrottleService service, String fieldName) throws Exception { Field field = VoteThrottleService.class.getDeclaredField(fieldName); field.setAccessible(true); - return ((Map) field.get(service)).size(); + return (Map) field.get(service); } } From dc9959264a3173fa5945ed2040a2de7d974f23e7 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:18:30 -0600 Subject: [PATCH 09/33] Address review feedback with regression coverage From 811dc1b4f13884f6054ae1ba4dbb5d2395f29a0e Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:18:41 -0600 Subject: [PATCH 10/33] Address review feedback with regression coverage --- .../java/com/vexsoftware/votifier/net/VoteThrottleService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 7796a20..13af84a 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -104,9 +104,9 @@ public String allowLog(String key, String msg) { long now = System.currentTimeMillis(); long windowMs = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; - trimLogStates(now); LogState state = logStates.get(key); if (state == null) { + trimLogStates(now); LogState created = new LogState(); LogState existing = logStates.putIfAbsent(key, created); state = existing == null ? created : existing; From e277ad52ec66f6971d7a0fd81c5a039fd5ddd910 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:18:52 -0600 Subject: [PATCH 11/33] Address review feedback with regression coverage From b8a5d7be07dd6b392c046430172b8f8c07f1b2ce Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:19:04 -0600 Subject: [PATCH 12/33] Address review feedback with regression coverage --- .../tests/VoteReceiverThrottleTest.java | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index e34cb3f..6c2df89 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -194,6 +194,20 @@ public void testExistingKeyDoesNotEvictAnotherStateAtCapacity() throws Exception assertEquals(4096, states.size()); } + @Test + public void testExistingLogKeyDoesNotEvictAnotherStateAtCapacity() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.allowLog("log:" + i, "message"); + } + Map states = stateMap(service, "logStates"); + Object existing = states.get("log:100"); + assertNull(service.allowLog("log:100", "message-again")); + assertTrue(existing == states.get("log:100")); + assertEquals(4096, states.size()); + } + private static int mapSize(VoteThrottleService service, String fieldName) throws Exception { return stateMap(service, fieldName).size(); } From de56202377e4fd034e7e6c7b570e0ca73e6ddea0 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:53:27 -0600 Subject: [PATCH 13/33] Address latest review feedback --- .../com/vexsoftware/votifier/net/VoteConnectionHandler.java | 6 ------ 1 file changed, 6 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index d95e2ca..5089d34 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -59,12 +59,6 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; tunnelMode = throttleService.isTunnelMode(remoteIp); - if (!tunnelMode && throttleService.isBlocked(throttleKey)) { - throttleService.logWarning(receiver, "throttle|" + throttleKey, - "Votifier rejected a throttled connection from " + remoteIp); - return null; - } - receiver.debug("Accepted connection from: " + address); accepted.setSoTimeout(5000); From af191c8236a4111e480c4fd0a4e19f300c576b75 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:53:45 -0600 Subject: [PATCH 14/33] Address latest review feedback --- .../votifier/net/VoteThrottleService.java | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 13af84a..ac52797 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -71,6 +71,9 @@ public void fail(String key, boolean tunnelMode, boolean realIpKnown) { long now = System.currentTimeMillis(); ThrottleState state = getThrottleState(key); + if (state == null) { + return; + } if (now - state.windowStartMs > config.windowMs) { state.windowStartMs = now; @@ -140,10 +143,12 @@ public void logSocketError(String remoteIp, SocketException ex) { public void logGenericError(String remoteIp, Exception ex) { } - private ThrottleState getThrottleState(String key) { + private synchronized ThrottleState getThrottleState(String key) { ThrottleState state = throttleStates.get(key); if (state == null) { - trimThrottleStates(System.currentTimeMillis()); + if (!trimThrottleStates(System.currentTimeMillis())) { + return null; + } ThrottleState created = new ThrottleState(); created.windowStartMs = System.currentTimeMillis(); ThrottleState existing = throttleStates.putIfAbsent(key, created); @@ -165,9 +170,9 @@ private void trimLogStates(long now) { removeOneIfFull(logStates); } - private void trimThrottleStates(long now) { + private boolean trimThrottleStates(long now) { if (throttleStates.size() < MAX_TRACKED_KEYS) { - return; + return true; } for (java.util.Map.Entry entry : throttleStates.entrySet()) { ThrottleState state = entry.getValue(); @@ -176,7 +181,17 @@ private void trimThrottleStates(long now) { throttleStates.remove(entry.getKey(), state); } } - removeOneIfFull(throttleStates); + if (throttleStates.size() < MAX_TRACKED_KEYS) { + return true; + } + for (java.util.Map.Entry entry : throttleStates.entrySet()) { + ThrottleState state = entry.getValue(); + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && throttleStates.remove(entry.getKey(), state)) { + return true; + } + } + return false; } private static void removeOneIfFull(ConcurrentHashMap states) { From 7d09b87df6b8e3499aa67bcd01ff10a69c663390 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:54:03 -0600 Subject: [PATCH 15/33] Address latest review feedback --- .../tests/VoteConnectionHandlerTest.java | 37 +------------------ 1 file changed, 2 insertions(+), 35 deletions(-) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java index c1916d9..c3251a8 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java @@ -260,42 +260,9 @@ public Vote call() { } @Test - public void testBlockedConnectionIsRejectedBeforeHandshake() throws Exception { + public void testProxyCanProvideClientIdentityBeforeThrottleDecision() throws Exception { receiver.setUseTokens(false); - ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "10s", 1, "30s", 1, "30s", - false, 999, "1s", "60s"); - VoteThrottleService throttleService = new VoteThrottleService(config); - VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); - - throttleService.fail("tunnel:127.0.0.1", false, false); - assertTrue(throttleService.isBlocked("tunnel:127.0.0.1")); - - try (ServerSocket serverSocket = new ServerSocket(0); - Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); - Socket accepted = serverSocket.accept()) { - - Future future = executor.submit(new Callable() { - @Override - public Vote call() { - return handler.handle(accepted); - } - }); - - BufferedReader clientReader = new BufferedReader( - new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); - - String handshake = clientReader.readLine(); - assertNull(handshake, "Blocked peers must not consume handshake or payload resources"); - - Vote vote = future.get(); - assertNull(vote); - } - } - - @Test - public void testConfiguredTunnelCanProvideClientIdentityBeforeThrottleDecision() throws Exception { - receiver.setUseTokens(false); - ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("127.0.0.1"), "10s", 1, "30s", 1, + ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "10s", 1, "30s", 1, "30s", false, 999, "1s", "60s"); VoteThrottleService throttleService = new VoteThrottleService(config); VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); From f7675a6d5f94f74e0c90214d787df29a65b754b2 Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 7 Sep 2026 17:54:20 -0600 Subject: [PATCH 16/33] Address latest review feedback --- .../tests/VoteReceiverThrottleTest.java | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 6c2df89..64c3d10 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -208,6 +208,19 @@ public void testExistingLogKeyDoesNotEvictAnotherStateAtCapacity() throws Except assertEquals(4096, states.size()); } + @Test + public void testOverflowDoesNotEvictActiveBan() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 1, "10s", 1, "10s", true, 1, "60s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + assertTrue(service.isBlocked("ip:100")); + service.fail("overflow", false, true); + assertTrue(service.isBlocked("ip:100")); + assertEquals(4096, mapSize(service, "throttleStates")); + } + private static int mapSize(VoteThrottleService service, String fieldName) throws Exception { return stateMap(service, fieldName).size(); } From 07d3fdf9c0dd7b36313686c6ff35032ce6bfef9b Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 19:00:16 -0600 Subject: [PATCH 17/33] Make vote throttling state updates atomic --- VotifierPlus/pom.xml | 13 ++- .../vexsoftware/votifier/net/VoteParser.java | 15 ++- .../votifier/net/VoteThrottleService.java | 92 ++++++++++--------- .../votifierplus/tests/VoteReceiverTest.java | 42 +++++++++ .../tests/VoteReceiverThrottleTest.java | 85 ++++++++++++++++- 5 files changed, 197 insertions(+), 50 deletions(-) diff --git a/VotifierPlus/pom.xml b/VotifierPlus/pom.xml index 4f3d41d..5343678 100644 --- a/VotifierPlus/pom.xml +++ b/VotifierPlus/pom.xml @@ -72,9 +72,14 @@ - - - org.apache.maven.plugins + + + org.apache.maven.plugins + maven-surefire-plugin + 3.5.4 + + + org.apache.maven.plugins maven-jar-plugin 3.5.0 @@ -485,4 +490,4 @@ - \ No newline at end of file + diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java index 61d8f41..526bf07 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteParser.java @@ -538,7 +538,8 @@ private VoteRequest parseV2(byte[] data, VoteReceiver receiver, String address, String serviceName = requireString(votePayload, FIELD_SERVICE_NAME, "Inner JSON from " + address + ": "); String username = requireString(votePayload, FIELD_USERNAME, "Inner JSON from " + address + ": "); - String voteAddress = votePayload.get(FIELD_ADDRESS).getAsString(); + String voteAddress = requirePossiblyEmptyString(votePayload, FIELD_ADDRESS, + "Inner JSON from " + address + ": "); String timeStamp = requireString(votePayload, FIELD_TIMESTAMP, "Inner JSON from " + address + ": "); String receivedChallenge = requireString(votePayload, FIELD_CHALLENGE, "Inner JSON from " + address + ": ").trim(); @@ -591,6 +592,18 @@ private String requireString(JsonObject obj, String field, String errorPrefix) t return value; } + private String requirePossiblyEmptyString(JsonObject obj, String field, String errorPrefix) + throws InvalidVoteException { + if (!obj.has(field)) { + throw new InvalidVoteException(errorPrefix + "missing field '" + field + "'"); + } + try { + return obj.get(field).getAsString(); + } catch (Exception ex) { + throw new InvalidVoteException(errorPrefix + "invalid field '" + field + "'", ex); + } + } + private String readString(byte[] data, int offset) { StringBuilder builder = new StringBuilder(); for (int i = offset; i < data.length; i++) { diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index ac52797..78a5de4 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -27,6 +27,8 @@ private static final class ThrottleState { private final ThrottleConfig config; private final ConcurrentHashMap logStates = new ConcurrentHashMap(); private final ConcurrentHashMap throttleStates = new ConcurrentHashMap(); + private final Object logStateLock = new Object(); + private final Object throttleStateLock = new Object(); public VoteThrottleService(ThrottleConfig config) { this.config = config; @@ -69,65 +71,69 @@ public void fail(String key, boolean tunnelMode, boolean realIpKnown) { return; } - long now = System.currentTimeMillis(); - ThrottleState state = getThrottleState(key); - if (state == null) { - return; - } + synchronized (throttleStateLock) { + long now = System.currentTimeMillis(); + ThrottleState state = getThrottleState(key); + if (state == null) { + return; + } - if (now - state.windowStartMs > config.windowMs) { - state.windowStartMs = now; - state.failures = 0; - } + if (now - state.windowStartMs > config.windowMs) { + state.windowStartMs = now; + state.failures = 0; + } - state.failures++; + state.failures++; - if (config.perClientBanEnabled && realIpKnown && state.failures >= config.perClientBanFailures) { - state.bannedUntilMs = now + config.perClientBanForMs; - return; - } + if (config.perClientBanEnabled && realIpKnown && state.failures >= config.perClientBanFailures) { + state.bannedUntilMs = now + config.perClientBanForMs; + return; + } - int threshold = tunnelMode ? config.tunnelFailures : config.failures; - long duration = tunnelMode ? config.tunnelThrottleForMs : config.throttleForMs; + int threshold = tunnelMode ? config.tunnelFailures : config.failures; + long duration = tunnelMode ? config.tunnelThrottleForMs : config.throttleForMs; - if (state.failures >= threshold) { - state.throttledUntilMs = now + duration; + if (state.failures >= threshold) { + state.throttledUntilMs = now + duration; + } } } public void success(String key) { - ThrottleState state = throttleStates.get(key); - if (state != null) { - state.failures = 0; - state.windowStartMs = System.currentTimeMillis(); + synchronized (throttleStateLock) { + ThrottleState state = throttleStates.get(key); + if (state != null) { + state.failures = 0; + state.windowStartMs = System.currentTimeMillis(); + } } } public String allowLog(String key, String msg) { - long now = System.currentTimeMillis(); - long windowMs = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; - - LogState state = logStates.get(key); - if (state == null) { - trimLogStates(now); - LogState created = new LogState(); - LogState existing = logStates.putIfAbsent(key, created); - state = existing == null ? created : existing; - } + synchronized (logStateLock) { + long now = System.currentTimeMillis(); + long windowMs = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; + LogState state = logStates.get(key); + if (state == null) { + trimLogStates(now); + state = new LogState(); + logStates.put(key, state); + } - if (now - state.lastLogMs >= windowMs) { - int suppressed = state.suppressed; - state.suppressed = 0; - state.lastLogMs = now; + if (now - state.lastLogMs >= windowMs) { + int suppressed = state.suppressed; + state.suppressed = 0; + state.lastLogMs = now; - if (suppressed > 0) { - return msg + " (suppressed " + suppressed + " similar in last " + windowMs + "ms)"; + if (suppressed > 0) { + return msg + " (suppressed " + suppressed + " similar in last " + windowMs + "ms)"; + } + return msg; } - return msg; - } - state.suppressed++; - return null; + state.suppressed++; + return null; + } } public void logWarning(VoteReceiver receiver, String key, String message) { @@ -143,7 +149,7 @@ public void logSocketError(String remoteIp, SocketException ex) { public void logGenericError(String remoteIp, Exception ex) { } - private synchronized ThrottleState getThrottleState(String key) { + private ThrottleState getThrottleState(String key) { ThrottleState state = throttleStates.get(key); if (state == null) { if (!trimThrottleStates(System.currentTimeMillis())) { diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java index fa11c82..4a34787 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverTest.java @@ -548,6 +548,30 @@ public void testV2VoteMissingUsernameField() throws Exception { assertTrue(exception.getMessage().contains("Missing required fields in inner JSON")); } + @Test + public void testV2VoteRejectsNullAddressAsInvalidVote() throws Exception { + JsonObject inner = validInnerVote(); + inner.add("address", null); + + InvalidVoteException exception = assertThrows(InvalidVoteException.class, + () -> parser.parse(v2Payload(inner), VoteProtocolVersion.V2, receiver, "test", + receiver.getChallenge())); + + assertTrue(exception.getMessage().contains("invalid field 'address'")); + } + + @Test + public void testV2VoteRejectsObjectAddressAsInvalidVote() throws Exception { + JsonObject inner = validInnerVote(); + inner.add("address", new JsonObject()); + + InvalidVoteException exception = assertThrows(InvalidVoteException.class, + () -> parser.parse(v2Payload(inner), VoteProtocolVersion.V2, receiver, "test", + receiver.getChallenge())); + + assertTrue(exception.getMessage().contains("invalid field 'address'")); + } + @Test public void testV2VoteInvalidChallenge() throws Exception { JsonObject inner = new JsonObject(); @@ -651,6 +675,24 @@ public void testBuildVoteFromParsedRequest() throws Exception { assertEquals("192.168.1.1", vote.getSourceAddress()); } + private JsonObject validInnerVote() { + JsonObject inner = new JsonObject(); + inner.addProperty("serviceName", "votifier.bencodez.com"); + inner.addProperty("username", "testUser"); + inner.addProperty("address", "127.0.0.1"); + inner.addProperty("timestamp", "TestTimestamp"); + inner.addProperty("challenge", receiver.getChallenge()); + return inner; + } + + private PushbackInputStream v2Payload(JsonObject inner) { + JsonObject outer = new JsonObject(); + outer.addProperty("payload", inner.toString()); + outer.addProperty("signature", "AA=="); + return new PushbackInputStream( + new ByteArrayInputStream(outer.toString().getBytes(StandardCharsets.UTF_8)), 512); + } + private byte[] createSignedV2Payload(String username) throws Exception { JsonObject inner = new JsonObject(); inner.addProperty("serviceName", "votifier.bencodez.com"); diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 64c3d10..6d43308 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -6,9 +6,13 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.lang.reflect.Field; import java.util.Collections; - import java.lang.reflect.Field; - import java.util.Map; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; @@ -180,6 +184,83 @@ public void testAttackerControlledKeysAreBounded() throws Exception { assertTrue(mapSize(service, "logStates") <= 4096); } + @Test + public void testConcurrentNewLogKeysStayWithinBound() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 2, "10s", false, 999, "1s")); + int workers = 16; + int keysPerWorker = 512; + ExecutorService executor = Executors.newFixedThreadPool(workers); + CountDownLatch ready = new CountDownLatch(workers); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(workers); + try { + for (int worker = 0; worker < workers; worker++) { + final int workerId = worker; + executor.execute(() -> { + ready.countDown(); + try { + start.await(); + for (int key = 0; key < keysPerWorker; key++) { + service.allowLog("concurrent-log:" + workerId + ':' + key, "message"); + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } finally { + finished.countDown(); + } + }); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + assertTrue(finished.await(10, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + assertEquals(4096, mapSize(service, "logStates")); + } + + @Test + public void testConcurrentFailuresUpdateStateAtomically() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("60s", 100000, "10s", 100000, "10s", false, 999, "1s")); + int workers = 16; + int failuresPerWorker = 256; + ExecutorService executor = Executors.newFixedThreadPool(workers); + CountDownLatch ready = new CountDownLatch(workers); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(workers); + try { + for (int worker = 0; worker < workers; worker++) { + executor.execute(() -> { + ready.countDown(); + try { + start.await(); + for (int failure = 0; failure < failuresPerWorker; failure++) { + service.fail("concurrent-failure", false, false); + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } finally { + finished.countDown(); + } + }); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + assertTrue(finished.await(10, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + Field statesField = VoteThrottleService.class.getDeclaredField("throttleStates"); + statesField.setAccessible(true); + Map states = (Map) statesField.get(service); + Object state = states.get("concurrent-failure"); + Field failuresField = state.getClass().getDeclaredField("failures"); + failuresField.setAccessible(true); + assertEquals(workers * failuresPerWorker, failuresField.getInt(state)); + } + @Test public void testExistingKeyDoesNotEvictAnotherStateAtCapacity() throws Exception { VoteThrottleService service = new VoteThrottleService( From 222fca73ea139d947a347423fe68744dd9a7fb29 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 19:59:20 -0600 Subject: [PATCH 18/33] Bound trusted tunnel throttle overflow --- .../votifier/net/VoteConnectionHandler.java | 14 +++--- .../votifier/net/VoteThrottleService.java | 50 +++++++++++++++++-- .../tests/VoteReceiverThrottleTest.java | 48 ++++++++++++++++++ 3 files changed, 103 insertions(+), 9 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index 5089d34..4294b13 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -46,6 +46,7 @@ public Vote handle(Socket socket) { String remoteIp = "unknown"; String address = ""; String throttleKey = null; + String aggregateThrottleKey = null; boolean tunnelMode = false; boolean realIpKnown = false; @@ -59,6 +60,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; tunnelMode = throttleService.isTunnelMode(remoteIp); + aggregateThrottleKey = tunnelMode ? throttleKey : null; receiver.debug("Accepted connection from: " + address); accepted.setSoTimeout(5000); @@ -77,8 +79,8 @@ public Vote handle(Socket socket) { realIpKnown = realIp != null && !realIp.isEmpty(); throttleKey = realIpKnown ? "ip:" + realIp : "tunnel:" + remoteIp; - if (throttleService.isBlocked(throttleKey)) { - long retry = throttleService.retryAfterMs(throttleKey); + if (throttleService.isBlocked(throttleKey, aggregateThrottleKey)) { + long retry = throttleService.retryAfterMs(throttleKey, aggregateThrottleKey); throttleService.logWarning(receiver, "throttle|" + throttleKey, "Votifier throttling " + throttleKey + " (tunnel=" + tunnelMode + "), retry in " + Math.max(0, retry / 1000) + "s"); return null; @@ -110,7 +112,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, realIpKnown); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, realIpKnown); throttleService.logWarning(receiver, "invalid|" + throttleKey, "Invalid vote format from " + remoteIp + ": " + ex.getMessage()); } catch (VoteAuthenticationException ex) { @@ -118,7 +120,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, realIpKnown); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, realIpKnown); throttleService.logWarning(receiver, "auth|" + throttleKey, "Authentication failed from " + remoteIp + ": " + ex.getMessage()); } catch (MalformedJsonException ex) { @@ -126,7 +128,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, false); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, false); throttleService.logWarning(receiver, "malformedjson|" + throttleKey, "Invalid vote format: Malformed JSON payload from " + remoteIp + " - " + ex.getMessage()); } catch (BadPaddingException ex) { @@ -134,7 +136,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; } - throttleService.fail(throttleKey, tunnelMode, realIpKnown); + throttleService.fail(throttleKey, aggregateThrottleKey, tunnelMode, realIpKnown); throttleService.logWarning(receiver, "badpadding|" + throttleKey, "Decryption failed: Invalid V1 vote block / public key mismatch from " + remoteIp); } catch (SocketTimeoutException ex) { diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 78a5de4..1d3dfcf 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -27,11 +27,19 @@ private static final class ThrottleState { private final ThrottleConfig config; private final ConcurrentHashMap logStates = new ConcurrentHashMap(); private final ConcurrentHashMap throttleStates = new ConcurrentHashMap(); + private final ConcurrentHashMap aggregateStates = + new ConcurrentHashMap(); private final Object logStateLock = new Object(); private final Object throttleStateLock = new Object(); + /* Configured tunnel remotes are finite, trusted aggregate identities. */ public VoteThrottleService(ThrottleConfig config) { this.config = config; + if (config != null) { + for (String remoteIp : config.tunnelRemoteIps) { + aggregateStates.put("tunnel:" + remoteIp, new ThrottleState()); + } + } } public ThrottleConfig getConfig() { @@ -43,30 +51,60 @@ public boolean isTunnelMode(String remoteIp) { } public boolean isBlocked(String key) { + return isBlocked(key, null); + } + + public boolean isBlocked(String key, String aggregateKey) { if (config == null || !config.enabled) { return false; } ThrottleState state = throttleStates.get(key); + if (isBlocked(state)) { + return true; + } + synchronized (throttleStateLock) { + if (aggregateKey != null && throttleStates.get(key) == null) + return isBlocked(aggregateStates.get(aggregateKey)); + } + return false; + } + + private boolean isBlocked(ThrottleState state) { if (state == null) { return false; } - long now = System.currentTimeMillis(); return state.bannedUntilMs > now || state.throttledUntilMs > now; } public long retryAfterMs(String key) { + return retryAfterMs(key, null); + } + + public long retryAfterMs(String key, String aggregateKey) { ThrottleState state = throttleStates.get(key); + long retry = retryAfterMs(state); + synchronized (throttleStateLock) { + if (aggregateKey != null && throttleStates.get(key) == null) + retry = Math.max(retry, retryAfterMs(aggregateStates.get(aggregateKey))); + } + return retry; + } + + private long retryAfterMs(ThrottleState state) { if (state == null) { return 0L; } - long now = System.currentTimeMillis(); return Math.max(state.bannedUntilMs, state.throttledUntilMs) - now; } public void fail(String key, boolean tunnelMode, boolean realIpKnown) { + fail(key, null, tunnelMode, realIpKnown); + } + + public void fail(String key, String aggregateKey, boolean tunnelMode, boolean realIpKnown) { if (config == null || !config.enabled) { return; } @@ -74,6 +112,11 @@ public void fail(String key, boolean tunnelMode, boolean realIpKnown) { synchronized (throttleStateLock) { long now = System.currentTimeMillis(); ThrottleState state = getThrottleState(key); + boolean aggregate = false; + if (state == null && aggregateKey != null) { + state = aggregateStates.get(aggregateKey); + aggregate = state != null; + } if (state == null) { return; } @@ -85,7 +128,8 @@ public void fail(String key, boolean tunnelMode, boolean realIpKnown) { state.failures++; - if (config.perClientBanEnabled && realIpKnown && state.failures >= config.perClientBanFailures) { + if (!aggregate && config.perClientBanEnabled && realIpKnown + && state.failures >= config.perClientBanFailures) { state.bannedUntilMs = now + config.perClientBanForMs; return; } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 6d43308..0ac9027 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -302,6 +302,54 @@ public void testOverflowDoesNotEvictActiveBan() throws Exception { assertEquals(4096, mapSize(service, "throttleStates")); } + @Test + public void testFullThrottleMapAccountsNewIdentityAgainstAggregateTunnel() throws Exception { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + String aggregateKey = "tunnel:proxy"; + String newIdentity = "ip:new"; + service.fail(newIdentity, aggregateKey, false, true); + + assertTrue(service.isBlocked(newIdentity, aggregateKey)); + assertTrue(service.retryAfterMs(newIdentity, aggregateKey) > 0); + assertEquals(4096, mapSize(service, "throttleStates")); + } + + @Test + public void testOverflowThrottleDoesNotBecomeGlobalAcrossTunnels() { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy-a", "proxy-b")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + service.fail("ip:new-a", "tunnel:proxy-a", false, true); + + assertTrue(service.isBlocked("ip:new-a", "tunnel:proxy-a")); + assertFalse(service.isBlocked("ip:new-b", "tunnel:proxy-b")); + } + + @Test + public void testOverflowFallbackIgnoresPrimaryTunnelState() { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy")); + service.fail("tunnel:proxy", true, false); + for (int i = 0; i < 4095; i++) { + service.fail("ip:" + i, false, true); + } + + assertFalse(service.isBlocked("ip:new", "tunnel:proxy")); + assertEquals(0L, service.retryAfterMs("ip:new", "tunnel:proxy")); + service.fail("ip:new", "tunnel:proxy", false, true); + assertTrue(service.isBlocked("ip:new", "tunnel:proxy")); + } + + private static ThrottleConfig tunnelCfg(String... remoteIps) { + return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, + "10s", 1, "10s", true, 1, "60s", "60s"); + } + private static int mapSize(VoteThrottleService service, String fieldName) throws Exception { return stateMap(service, fieldName).size(); } From cf37fe1e5327bddcdf9e584089ca0bd4942094c0 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 20:19:44 -0600 Subject: [PATCH 19/33] Preserve active throttle state at capacity --- .../votifier/net/VoteThrottleService.java | 7 ------- .../tests/VoteReceiverThrottleTest.java | 17 +++++++++++++++++ 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 1d3dfcf..e9c9e25 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -234,13 +234,6 @@ private boolean trimThrottleStates(long now) { if (throttleStates.size() < MAX_TRACKED_KEYS) { return true; } - for (java.util.Map.Entry entry : throttleStates.entrySet()) { - ThrottleState state = entry.getValue(); - if (state.bannedUntilMs <= now && state.throttledUntilMs <= now - && throttleStates.remove(entry.getKey(), state)) { - return true; - } - } return false; } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 0ac9027..b276532 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -318,6 +318,23 @@ public void testFullThrottleMapAccountsNewIdentityAgainstAggregateTunnel() throw assertEquals(4096, mapSize(service, "throttleStates")); } + @Test + public void testFullMapPreservesInWindowCountersAndUsesAggregateTunnel() throws Exception { + ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("proxy"), "5s", 2, "10s", 2, + "10s", false, 999, "1s", "60s"); + VoteThrottleService service = new VoteThrottleService(config); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + service.fail("ip:new", "tunnel:proxy", false, true); + assertFalse(service.isBlocked("ip:new", "tunnel:proxy")); + service.fail("ip:new", "tunnel:proxy", false, true); + + assertTrue(service.isBlocked("ip:new", "tunnel:proxy")); + assertEquals(4096, mapSize(service, "throttleStates")); + } + @Test public void testOverflowThrottleDoesNotBecomeGlobalAcrossTunnels() { VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy-a", "proxy-b")); From 10240faf7cb9fddd690210ee3d686bfcb99062af Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 20:45:14 -0600 Subject: [PATCH 20/33] Throttle overflow for non-tunnel proxy peers --- .../votifier/net/VoteConnectionHandler.java | 2 +- .../votifier/net/VoteThrottleService.java | 39 ++++++++++++++++++- .../tests/VoteConnectionHandlerTest.java | 38 ++++++++++++++++++ .../tests/VoteReceiverThrottleTest.java | 18 +++++++++ 4 files changed, 95 insertions(+), 2 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index 4294b13..4965a54 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -60,7 +60,7 @@ public Vote handle(Socket socket) { throttleKey = "tunnel:" + remoteIp; tunnelMode = throttleService.isTunnelMode(remoteIp); - aggregateThrottleKey = tunnelMode ? throttleKey : null; + aggregateThrottleKey = throttleKey; receiver.debug("Accepted connection from: " + address); accepted.setSoTimeout(5000); diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index e9c9e25..3903581 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -114,7 +114,7 @@ public void fail(String key, String aggregateKey, boolean tunnelMode, boolean re ThrottleState state = getThrottleState(key); boolean aggregate = false; if (state == null && aggregateKey != null) { - state = aggregateStates.get(aggregateKey); + state = getAggregateThrottleState(aggregateKey, now); aggregate = state != null; } if (state == null) { @@ -207,6 +207,21 @@ private ThrottleState getThrottleState(String key) { return state; } + private ThrottleState getAggregateThrottleState(String key, long now) { + ThrottleState state = aggregateStates.get(key); + if (state != null) { + return state; + } + if (!trimAggregateStates(now)) { + return null; + } + + ThrottleState created = new ThrottleState(); + created.windowStartMs = now; + ThrottleState existing = aggregateStates.putIfAbsent(key, created); + return existing == null ? created : existing; + } + private void trimLogStates(long now) { if (logStates.size() < MAX_TRACKED_KEYS) { return; @@ -237,6 +252,28 @@ private boolean trimThrottleStates(long now) { return false; } + private boolean trimAggregateStates(long now) { + if (aggregateStates.size() < MAX_TRACKED_KEYS) { + return true; + } + for (java.util.Map.Entry entry : aggregateStates.entrySet()) { + if (isConfiguredAggregateKey(entry.getKey())) { + continue; + } + ThrottleState state = entry.getValue(); + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && now - state.windowStartMs > config.windowMs) { + aggregateStates.remove(entry.getKey(), state); + } + } + return aggregateStates.size() < MAX_TRACKED_KEYS; + } + + private boolean isConfiguredAggregateKey(String key) { + return config != null && key.startsWith("tunnel:") + && config.tunnelRemoteIps.contains(key.substring("tunnel:".length())); + } + private static void removeOneIfFull(ConcurrentHashMap states) { if (states.size() >= MAX_TRACKED_KEYS) { java.util.Iterator iterator = states.keySet().iterator(); diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java index c3251a8..aac78ef 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java @@ -442,6 +442,44 @@ public Vote call() { } } + @Test + public void testNonTunnelProxyOverflowUsesRemoteAggregate() throws Exception { + receiver.setUseTokens(false); + ThrottleConfig config = new ThrottleConfig(true, Collections.emptySet(), "5s", 1, "10s", 1, + "10s", false, 999, "1s", "60s"); + VoteThrottleService throttleService = new VoteThrottleService(config); + for (int i = 0; i < 4096; i++) { + throttleService.fail("ip:" + i, false, true); + } + VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); + + try (ServerSocket serverSocket = new ServerSocket(0); + Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); + Socket accepted = serverSocket.accept()) { + + Future future = executor.submit(new Callable() { + @Override + public Vote call() { + return handler.handle(accepted); + } + }); + + BufferedReader clientReader = new BufferedReader( + new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); + OutputStream clientOut = client.getOutputStream(); + assertEquals("VOTIFIER 1", clientReader.readLine()); + + clientOut.write("PROXY TCP4 203.0.113.10 127.0.0.1 1234 8192\r\n" + .getBytes(StandardCharsets.US_ASCII)); + clientOut.write(new byte[256]); + clientOut.flush(); + client.shutdownOutput(); + + assertNull(future.get()); + assertTrue(throttleService.isBlocked("ip:203.0.113.10", "tunnel:127.0.0.1")); + } + } + @Test public void testHandleTestVoteDoesNotSendOkResponse() throws Exception { receiver.setUseTokens(false); diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index b276532..4e8256b 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -335,6 +335,24 @@ public void testFullMapPreservesInWindowCountersAndUsesAggregateTunnel() throws assertEquals(4096, mapSize(service, "throttleStates")); } + @Test + public void testFullThrottleMapAccountsNewIdentityAgainstNonTunnelAggregate() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 1, "10s", 1, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, true); + } + + String aggregateKey = "tunnel:proxy"; + String newIdentity = "ip:new"; + service.fail(newIdentity, aggregateKey, false, true); + + assertTrue(service.isBlocked(newIdentity, aggregateKey)); + assertTrue(service.retryAfterMs(newIdentity, aggregateKey) > 0); + assertEquals(4096, mapSize(service, "throttleStates")); + assertEquals(1, mapSize(service, "aggregateStates")); + } + @Test public void testOverflowThrottleDoesNotBecomeGlobalAcrossTunnels() { VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy-a", "proxy-b")); From fc7bb48d88d232831085abe816478bd9c2dffc99 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 21:38:49 -0600 Subject: [PATCH 21/33] Harden aggregate throttle overflow --- .../votifier/net/VoteConnectionHandler.java | 5 +- .../votifier/net/VoteThrottleService.java | 53 ++++++++++++++-- .../tests/VoteReceiverThrottleTest.java | 60 +++++++++++++++++++ 3 files changed, 112 insertions(+), 6 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index 4965a54..94dcbb0 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -81,7 +81,8 @@ public Vote handle(Socket socket) { if (throttleService.isBlocked(throttleKey, aggregateThrottleKey)) { long retry = throttleService.retryAfterMs(throttleKey, aggregateThrottleKey); - throttleService.logWarning(receiver, "throttle|" + throttleKey, "Votifier throttling " + throttleKey + String blockedKey = throttleService.blockedKey(throttleKey, aggregateThrottleKey); + throttleService.logWarning(receiver, "throttle|" + blockedKey, "Votifier throttling " + blockedKey + " (tunnel=" + tunnelMode + "), retry in " + Math.max(0, retry / 1000) + "s"); return null; } @@ -100,7 +101,7 @@ public Vote handle(Socket socket) { } receiver.log("Received vote record -> " + vote); - throttleService.success(throttleKey); + throttleService.success(throttleKey, aggregateThrottleKey); if (!"TestVote".equalsIgnoreCase(vote.getTimeStamp())) { sendOkResponse(writer); diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 3903581..2189af9 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -11,6 +11,7 @@ public class VoteThrottleService { private static final int MAX_TRACKED_KEYS = 4096; + private static final int AGGREGATE_OVERFLOW_BUCKETS = 64; private static final class LogState { private volatile long lastLogMs; @@ -28,7 +29,8 @@ private static final class ThrottleState { private final ConcurrentHashMap logStates = new ConcurrentHashMap(); private final ConcurrentHashMap throttleStates = new ConcurrentHashMap(); private final ConcurrentHashMap aggregateStates = - new ConcurrentHashMap(); + new ConcurrentHashMap(); + private final ThrottleState[] aggregateOverflowStates = new ThrottleState[AGGREGATE_OVERFLOW_BUCKETS]; private final Object logStateLock = new Object(); private final Object throttleStateLock = new Object(); /* Configured tunnel remotes are finite, trusted aggregate identities. */ @@ -65,11 +67,27 @@ public boolean isBlocked(String key, String aggregateKey) { } synchronized (throttleStateLock) { if (aggregateKey != null && throttleStates.get(key) == null) - return isBlocked(aggregateStates.get(aggregateKey)); + return isBlocked(getAggregateState(aggregateKey)); } return false; } + public String blockedKey(String key, String aggregateKey) { + if (config == null || !config.enabled) { + return key; + } + synchronized (throttleStateLock) { + ThrottleState direct = throttleStates.get(key); + if (isBlocked(direct)) { + return key; + } + if (aggregateKey != null && direct == null && isBlocked(getAggregateState(aggregateKey))) { + return aggregateKey; + } + return key; + } + } + private boolean isBlocked(ThrottleState state) { if (state == null) { return false; @@ -87,7 +105,7 @@ public long retryAfterMs(String key, String aggregateKey) { long retry = retryAfterMs(state); synchronized (throttleStateLock) { if (aggregateKey != null && throttleStates.get(key) == null) - retry = Math.max(retry, retryAfterMs(aggregateStates.get(aggregateKey))); + retry = Math.max(retry, retryAfterMs(getAggregateState(aggregateKey))); } return retry; } @@ -144,8 +162,15 @@ public void fail(String key, String aggregateKey, boolean tunnelMode, boolean re } public void success(String key) { + success(key, null); + } + + public void success(String key, String aggregateKey) { synchronized (throttleStateLock) { ThrottleState state = throttleStates.get(key); + if (state == null && aggregateKey != null) { + state = getAggregateState(aggregateKey); + } if (state != null) { state.failures = 0; state.windowStartMs = System.currentTimeMillis(); @@ -213,7 +238,7 @@ private ThrottleState getAggregateThrottleState(String key, long now) { return state; } if (!trimAggregateStates(now)) { - return null; + return getOverflowAggregateState(key, now); } ThrottleState created = new ThrottleState(); @@ -222,6 +247,26 @@ private ThrottleState getAggregateThrottleState(String key, long now) { return existing == null ? created : existing; } + private ThrottleState getAggregateState(String key) { + ThrottleState state = aggregateStates.get(key); + if (state != null) { + return state; + } + return aggregateStates.size() >= MAX_TRACKED_KEYS + ? aggregateOverflowStates[Math.floorMod(key.hashCode(), AGGREGATE_OVERFLOW_BUCKETS)] : null; + } + + private ThrottleState getOverflowAggregateState(String key, long now) { + int index = Math.floorMod(key.hashCode(), AGGREGATE_OVERFLOW_BUCKETS); + ThrottleState state = aggregateOverflowStates[index]; + if (state == null) { + state = new ThrottleState(); + state.windowStartMs = now; + aggregateOverflowStates[index] = state; + } + return state; + } + private void trimLogStates(long now) { if (logStates.size() < MAX_TRACKED_KEYS) { return; diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 4e8256b..ffac2f3 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -143,6 +143,20 @@ public void testSuccessResetsFailureCounter() { assertTrue(service.isBlocked(key)); } + @Test + public void testSuccessResetsOverflowAggregateForDirectPeer() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + String key = "tunnel:direct"; + service.fail(key, key, false, false); + service.success(key, key); + service.fail(key, key, false, false); + assertFalse(service.isBlocked(key, key)); + } + @Test public void testWindowExpiryResetsFailureCounter() throws Exception { VoteThrottleService service = new VoteThrottleService( @@ -318,6 +332,18 @@ public void testFullThrottleMapAccountsNewIdentityAgainstAggregateTunnel() throw assertEquals(4096, mapSize(service, "throttleStates")); } + @Test + public void testAggregateBlockUsesAggregateLogKey() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 1, "10s", 1, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + String aggregateKey = "tunnel:proxy"; + service.fail("ip:rotated", aggregateKey, false, false); + assertEquals(aggregateKey, service.blockedKey("ip:another", aggregateKey)); + } + @Test public void testFullMapPreservesInWindowCountersAndUsesAggregateTunnel() throws Exception { ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("proxy"), "5s", 2, "10s", 2, @@ -380,6 +406,40 @@ public void testOverflowFallbackIgnoresPrimaryTunnelState() { assertTrue(service.isBlocked("ip:new", "tunnel:proxy")); } + @Test + public void testAggregateCapacityUsesBoundedOverflowBuckets() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + for (int i = 0; i < 4096; i++) { + service.fail("ip:overflow:" + i, "remote:" + i, false, false); + } + String aggregateKey = "remote:overflow"; + service.fail("ip:new", aggregateKey, false, false); + service.fail("ip:new", aggregateKey, false, false); + assertTrue(service.isBlocked("ip:new", aggregateKey)); + assertEquals(4096, mapSize(service, "aggregateStates")); + } + + @Test + public void testOverflowBucketIsIgnoredAfterAggregateCapacityRecovers() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) service.fail("ip:" + i, false, false); + for (int i = 0; i < 4096; i++) service.fail("ip:overflow:" + i, "remote:" + i, false, false); + assertEquals("Aa".hashCode(), "BB".hashCode()); + service.fail("ip:overflow-a", "Aa", false, false); + service.fail("ip:overflow-a", "Aa", false, false); + assertTrue(service.isBlocked("ip:overflow-b", "BB")); + + Map aggregates = stateMap(service, "aggregateStates"); + aggregates.remove(aggregates.keySet().iterator().next()); + assertFalse(service.isBlocked("ip:overflow-b", "BB"), + "stale overflow state must not shadow a newly available aggregate slot"); + } + private static ThrottleConfig tunnelCfg(String... remoteIps) { return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, "10s", 1, "10s", true, 1, "60s", "60s"); From 570e340109db0768816bf14dbf371179506a0e8c Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 22:38:25 -0600 Subject: [PATCH 22/33] fix: enforce shared proxy throttles --- .../votifier/net/VoteThrottleService.java | 17 ++++++---- .../tests/VoteReceiverThrottleTest.java | 34 +++++++++++++++++++ 2 files changed, 45 insertions(+), 6 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 2189af9..afa2bfe 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -66,7 +66,7 @@ public boolean isBlocked(String key, String aggregateKey) { return true; } synchronized (throttleStateLock) { - if (aggregateKey != null && throttleStates.get(key) == null) + if (aggregateKey != null) return isBlocked(getAggregateState(aggregateKey)); } return false; @@ -81,7 +81,7 @@ public String blockedKey(String key, String aggregateKey) { if (isBlocked(direct)) { return key; } - if (aggregateKey != null && direct == null && isBlocked(getAggregateState(aggregateKey))) { + if (aggregateKey != null && isBlocked(getAggregateState(aggregateKey))) { return aggregateKey; } return key; @@ -104,7 +104,7 @@ public long retryAfterMs(String key, String aggregateKey) { ThrottleState state = throttleStates.get(key); long retry = retryAfterMs(state); synchronized (throttleStateLock) { - if (aggregateKey != null && throttleStates.get(key) == null) + if (aggregateKey != null) retry = Math.max(retry, retryAfterMs(getAggregateState(aggregateKey))); } return retry; @@ -168,13 +168,18 @@ public void success(String key) { public void success(String key, String aggregateKey) { synchronized (throttleStateLock) { ThrottleState state = throttleStates.get(key); - if (state == null && aggregateKey != null) { - state = getAggregateState(aggregateKey); - } if (state != null) { state.failures = 0; state.windowStartMs = System.currentTimeMillis(); } + /* A proxied success must not clear failures shared by other identities. */ + if (aggregateKey != null && key.equals(aggregateKey)) { + state = getAggregateState(aggregateKey); + if (state != null) { + state.failures = 0; + state.windowStartMs = System.currentTimeMillis(); + } + } } } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index ffac2f3..e5d4665 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -344,6 +344,40 @@ public void testAggregateBlockUsesAggregateLogKey() { assertEquals(aggregateKey, service.blockedKey("ip:another", aggregateKey)); } + @Test + public void testAggregateBlockAppliesToExistingPrimaryIdentity() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + String primaryKey = "ip:existing"; + String aggregateKey = "tunnel:proxy"; + service.fail(primaryKey, false, false); + for (int i = 0; i < 4095; i++) { + service.fail("ip:filler:" + i, false, false); + } + service.fail("ip:proxied", aggregateKey, false, false); + service.fail("ip:proxied", aggregateKey, false, false); + + assertTrue(service.isBlocked(primaryKey, aggregateKey)); + assertEquals(aggregateKey, service.blockedKey(primaryKey, aggregateKey)); + assertTrue(service.retryAfterMs(primaryKey, aggregateKey) > 0); + } + + @Test + public void testProxiedSuccessDoesNotClearSharedAggregateFailures() { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + String aggregateKey = "tunnel:proxy"; + for (int i = 0; i < 4096; i++) { + service.fail("ip:filler:" + i, false, false); + } + service.fail("ip:proxied", aggregateKey, false, false); + service.success("ip:proxied", aggregateKey); + service.fail("ip:rotated", aggregateKey, false, false); + + assertTrue(service.isBlocked("ip:another", aggregateKey), + "a proxied success must not erase another identity's aggregate failure"); + } + @Test public void testFullMapPreservesInWindowCountersAndUsesAggregateTunnel() throws Exception { ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("proxy"), "5s", 2, "10s", 2, From f04f46dd1a924393c6ab59ac4a2238b67f8b9fde Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Mon, 7 Sep 2026 23:11:35 -0600 Subject: [PATCH 23/33] Preserve shared overflow throttle failures --- .../votifier/net/VoteThrottleService.java | 5 ++++- .../tests/VoteReceiverThrottleTest.java | 15 +++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index afa2bfe..cf7ae9c 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -174,7 +174,10 @@ public void success(String key, String aggregateKey) { } /* A proxied success must not clear failures shared by other identities. */ if (aggregateKey != null && key.equals(aggregateKey)) { - state = getAggregateState(aggregateKey); + // Only a dedicated aggregate belongs to this identity. An overflow + // bucket is deliberately shared by many identities and must not be + // reset by one successful request. + state = aggregateStates.get(aggregateKey); if (state != null) { state.failures = 0; state.windowStartMs = System.currentTimeMillis(); diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index e5d4665..67aff09 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -474,6 +474,21 @@ public void testOverflowBucketIsIgnoredAfterAggregateCapacityRecovers() throws E "stale overflow state must not shadow a newly available aggregate slot"); } + @Test + public void testDirectSuccessDoesNotResetSharedAggregateOverflowBucket() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) service.fail("ip:" + i, false, false); + for (int i = 0; i < 4096; i++) service.fail("ip:overflow:" + i, "remote:" + i, false, false); + assertEquals("Aa".hashCode(), "BB".hashCode()); + service.fail("ip:overflow-a", "Aa", false, false); + service.success("Aa", "Aa"); + service.fail("ip:overflow-b", "BB", false, false); + + assertTrue(service.isBlocked("ip:overflow-c", "BB"), + "a direct success must not erase failures belonging to a shared overflow bucket"); + } + private static ThrottleConfig tunnelCfg(String... remoteIps) { return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, "10s", 1, "10s", true, 1, "60s", "60s"); From afd73d9b16edf0773bb11505208b4c3a538b89b6 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 00:15:01 -0600 Subject: [PATCH 24/33] Preserve aggregate throttle failure windows --- .../votifier/net/VoteThrottleService.java | 17 +++++++++++++++- .../tests/VoteReceiverThrottleTest.java | 20 +++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index cf7ae9c..0d6975b 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -129,8 +129,18 @@ public void fail(String key, String aggregateKey, boolean tunnelMode, boolean re synchronized (throttleStateLock) { long now = System.currentTimeMillis(); - ThrottleState state = getThrottleState(key); boolean aggregate = false; + ThrottleState state = throttleStates.get(key); + if (state == null && aggregateKey != null) { + ThrottleState existingAggregate = getAggregateState(aggregateKey); + if (hasActiveFailures(existingAggregate, now)) { + state = existingAggregate; + aggregate = true; + } + } + if (state == null) { + state = getThrottleState(key); + } if (state == null && aggregateKey != null) { state = getAggregateThrottleState(aggregateKey, now); aggregate = state != null; @@ -161,6 +171,11 @@ public void fail(String key, String aggregateKey, boolean tunnelMode, boolean re } } + private boolean hasActiveFailures(ThrottleState state, long now) { + return state != null && (state.bannedUntilMs > now || state.throttledUntilMs > now + || state.failures > 0 && now - state.windowStartMs <= config.windowMs); + } + public void success(String key) { success(key, null); } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 67aff09..f2f4f47 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -332,6 +332,26 @@ public void testFullThrottleMapAccountsNewIdentityAgainstAggregateTunnel() throw assertEquals(4096, mapSize(service, "throttleStates")); } + @Test + public void testAggregateFallbackRemainsBoundAfterPrimaryCapacityRecovers() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 3, "10s", false, 999, "1s")); + String aggregateKey = "tunnel:proxy"; + for (int i = 0; i < 4096; i++) { + service.fail("ip:" + i, false, false); + } + + service.fail("ip:reused", aggregateKey, false, false); + service.fail("ip:reused", aggregateKey, false, false); + stateMap(service, "throttleStates").remove("ip:0"); + + service.fail("ip:reused", aggregateKey, false, false); + + assertTrue(service.isBlocked("ip:reused", aggregateKey), + "an identity that used aggregate fallback must not split its failure counter after capacity recovers"); + assertFalse(stateMap(service, "throttleStates").containsKey("ip:reused")); + } + @Test public void testAggregateBlockUsesAggregateLogKey() { VoteThrottleService service = new VoteThrottleService( From 13798d785da458b420cc7d4eca8b6849f0a138a4 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 00:53:29 -0600 Subject: [PATCH 25/33] Recheck throttles under state lock --- .../votifier/net/VoteThrottleService.java | 6 ++--- .../tests/VoteReceiverThrottleTest.java | 25 +++++++++++++++++++ 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 0d6975b..feb3594 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -61,11 +61,9 @@ public boolean isBlocked(String key, String aggregateKey) { return false; } - ThrottleState state = throttleStates.get(key); - if (isBlocked(state)) { - return true; - } synchronized (throttleStateLock) { + ThrottleState state = throttleStates.get(key); + if (isBlocked(state)) return true; if (aggregateKey != null) return isBlocked(getAggregateState(aggregateKey)); } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index f2f4f47..6c09c63 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -13,6 +13,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; @@ -509,6 +510,30 @@ public void testDirectSuccessDoesNotResetSharedAggregateOverflowBucket() throws "a direct success must not erase failures belonging to a shared overflow bucket"); } + @Test + public void testBlockedCheckRechecksPrimaryAfterWaitingForFailureUpdate() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + String key = "ip:concurrent"; + service.fail(key, false, true); + AtomicBoolean blocked = new AtomicBoolean(); + Thread check = new Thread(() -> blocked.set(service.isBlocked(key, "tunnel:other"))); + Field lockField = VoteThrottleService.class.getDeclaredField("throttleStateLock"); + lockField.setAccessible(true); + Object lock = lockField.get(service); + + synchronized (lock) { + check.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); + while (check.getState() != Thread.State.BLOCKED && System.nanoTime() < deadline) Thread.onSpinWait(); + assertEquals(Thread.State.BLOCKED, check.getState()); + service.fail(key, false, true); + } + check.join(TimeUnit.SECONDS.toMillis(2)); + + assertTrue(blocked.get(), "the primary throttle applied while waiting must not be bypassed"); + } + private static ThrottleConfig tunnelCfg(String... remoteIps) { return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, "10s", 1, "10s", true, 1, "60s", "60s"); From 18476548e8c812d7199709b2a2c78a0e5ab53da4 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 01:16:55 -0600 Subject: [PATCH 26/33] Bound saturated throttle lookup work --- .../votifier/net/VoteThrottleService.java | 27 ++++++++++++++++--- .../tests/VoteReceiverThrottleTest.java | 23 ++++++++++++++++ 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index feb3594..29e62a8 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -33,6 +33,9 @@ private static final class ThrottleState { private final ThrottleState[] aggregateOverflowStates = new ThrottleState[AGGREGATE_OVERFLOW_BUCKETS]; private final Object logStateLock = new Object(); private final Object throttleStateLock = new Object(); + /* Guarded by throttleStateLock; avoid O(n) sweeps while all entries are known active. */ + private long nextThrottleSweepMs; + private long nextAggregateSweepMs; /* Configured tunnel remotes are finite, trusted aggregate identities. */ public VoteThrottleService(ThrottleConfig config) { @@ -80,7 +83,9 @@ public String blockedKey(String key, String aggregateKey) { return key; } if (aggregateKey != null && isBlocked(getAggregateState(aggregateKey))) { - return aggregateKey; + if (aggregateStates.containsKey(aggregateKey)) return aggregateKey; + return "aggregate-overflow:" + + Math.floorMod(aggregateKey.hashCode(), AGGREGATE_OVERFLOW_BUCKETS); } return key; } @@ -305,16 +310,20 @@ private boolean trimThrottleStates(long now) { if (throttleStates.size() < MAX_TRACKED_KEYS) { return true; } + if (now < nextThrottleSweepMs) return false; + long nextSweep = Long.MAX_VALUE; for (java.util.Map.Entry entry : throttleStates.entrySet()) { ThrottleState state = entry.getValue(); if (state.bannedUntilMs <= now && state.throttledUntilMs <= now && now - state.windowStartMs > config.windowMs) { throttleStates.remove(entry.getKey(), state); - } + } else nextSweep = Math.min(nextSweep, stateExpiry(state)); } if (throttleStates.size() < MAX_TRACKED_KEYS) { + nextThrottleSweepMs = 0L; return true; } + nextThrottleSweepMs = nextSweep; return false; } @@ -322,6 +331,8 @@ private boolean trimAggregateStates(long now) { if (aggregateStates.size() < MAX_TRACKED_KEYS) { return true; } + if (now < nextAggregateSweepMs) return false; + long nextSweep = Long.MAX_VALUE; for (java.util.Map.Entry entry : aggregateStates.entrySet()) { if (isConfiguredAggregateKey(entry.getKey())) { continue; @@ -330,9 +341,17 @@ private boolean trimAggregateStates(long now) { if (state.bannedUntilMs <= now && state.throttledUntilMs <= now && now - state.windowStartMs > config.windowMs) { aggregateStates.remove(entry.getKey(), state); - } + } else nextSweep = Math.min(nextSweep, stateExpiry(state)); } - return aggregateStates.size() < MAX_TRACKED_KEYS; + boolean available = aggregateStates.size() < MAX_TRACKED_KEYS; + nextAggregateSweepMs = available ? 0L : nextSweep; + return available; + } + + private long stateExpiry(ThrottleState state) { + long windowExpiry = state.windowStartMs > Long.MAX_VALUE - config.windowMs - 1L + ? Long.MAX_VALUE : state.windowStartMs + config.windowMs + 1L; + return Math.max(windowExpiry, Math.max(state.bannedUntilMs, state.throttledUntilMs)); } private boolean isConfiguredAggregateKey(String key) { diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 6c09c63..111467b 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -488,6 +488,9 @@ public void testOverflowBucketIsIgnoredAfterAggregateCapacityRecovers() throws E service.fail("ip:overflow-a", "Aa", false, false); service.fail("ip:overflow-a", "Aa", false, false); assertTrue(service.isBlocked("ip:overflow-b", "BB")); + assertEquals("aggregate-overflow:" + Math.floorMod("BB".hashCode(), 64), + service.blockedKey("ip:overflow-b", "BB"), + "all remotes rejected by one overflow bucket must share its log key"); Map aggregates = stateMap(service, "aggregateStates"); aggregates.remove(aggregates.keySet().iterator().next()); @@ -534,6 +537,20 @@ public void testBlockedCheckRechecksPrimaryAfterWaitingForFailureUpdate() throws assertTrue(blocked.get(), "the primary throttle applied while waiting must not be bypassed"); } + @Test + public void testSaturatedPrimaryStateCachesItsNextPossibleSweep() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:active:" + index, false, true); + + service.fail("ip:overflow:first", false, true); + long nextSweep = longField(service, "nextThrottleSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + service.fail("ip:overflow:second", false, true); + assertEquals(nextSweep, longField(service, "nextThrottleSweepMs"), + "a miss before the earliest expiry must reuse the saturated-state decision"); + } + private static ThrottleConfig tunnelCfg(String... remoteIps) { return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, "10s", 1, "10s", true, 1, "60s", "60s"); @@ -548,4 +565,10 @@ private static int mapSize(VoteThrottleService service, String fieldName) throws field.setAccessible(true); return (Map) field.get(service); } + + private static long longField(VoteThrottleService service, String fieldName) throws Exception { + Field field = VoteThrottleService.class.getDeclaredField(fieldName); + field.setAccessible(true); + return field.getLong(service); + } } From 7aa45c38716ffc2aca7c07e933708017d7753f59 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 02:16:41 -0600 Subject: [PATCH 27/33] Reclaim inactive throttle states promptly --- .../votifier/net/VoteThrottleService.java | 13 +++++++++++-- .../tests/VoteReceiverThrottleTest.java | 17 +++++++++++++++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 29e62a8..1a5c2b6 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -185,10 +185,15 @@ public void success(String key) { public void success(String key, String aggregateKey) { synchronized (throttleStateLock) { + long now = System.currentTimeMillis(); ThrottleState state = throttleStates.get(key); if (state != null) { state.failures = 0; - state.windowStartMs = System.currentTimeMillis(); + state.windowStartMs = now; + if (state.bannedUntilMs <= now && state.throttledUntilMs <= now + && throttleStates.remove(key, state)) { + nextThrottleSweepMs = 0L; + } } /* A proxied success must not clear failures shared by other identities. */ if (aggregateKey != null && key.equals(aggregateKey)) { @@ -198,7 +203,11 @@ public void success(String key, String aggregateKey) { state = aggregateStates.get(aggregateKey); if (state != null) { state.failures = 0; - state.windowStartMs = System.currentTimeMillis(); + state.windowStartMs = now; + if (!isConfiguredAggregateKey(aggregateKey) && state.bannedUntilMs <= now + && state.throttledUntilMs <= now && aggregateStates.remove(aggregateKey, state)) { + nextAggregateSweepMs = 0L; + } } } } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 111467b..ecfa977 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -144,6 +144,23 @@ public void testSuccessResetsFailureCounter() { assertTrue(service.isBlocked(key)); } + @Test + public void testSuccessImmediatelyReclaimsPrimaryCapacity() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 3, "10s", 3, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:active:" + index, false, true); + + service.fail("ip:overflow", false, true); + assertTrue(longField(service, "nextThrottleSweepMs") > System.currentTimeMillis()); + service.success("ip:active:0"); + + assertEquals(4095, mapSize(service, "throttleStates")); + assertEquals(0L, longField(service, "nextThrottleSweepMs")); + service.fail("ip:replacement", false, true); + assertTrue(stateMap(service, "throttleStates").containsKey("ip:replacement"), + "a successful identity must release its primary slot immediately"); + } + @Test public void testSuccessResetsOverflowAggregateForDirectPeer() { VoteThrottleService service = new VoteThrottleService( From fdb41445642801863ec6fd436738f7ce672edaed Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 02:36:15 -0600 Subject: [PATCH 28/33] Reject aggregate-blocked peers before payload --- .../votifier/net/VoteConnectionHandler.java | 9 +++++ .../votifier/net/VoteThrottleService.java | 35 +++++++++++++++++++ .../tests/VoteConnectionHandlerTest.java | 31 ++++++++++++++++ .../tests/VoteReceiverThrottleTest.java | 3 ++ 4 files changed, 78 insertions(+) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java index 94dcbb0..47343d9 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteConnectionHandler.java @@ -64,6 +64,15 @@ public Vote handle(Socket socket) { receiver.debug("Accepted connection from: " + address); accepted.setSoTimeout(5000); + if (throttleService.isAggregateBlocked(aggregateThrottleKey)) { + long retry = throttleService.aggregateRetryAfterMs(aggregateThrottleKey); + String blockedKey = throttleService.aggregateBlockedKey(aggregateThrottleKey); + throttleService.logWarning(receiver, "throttle|" + blockedKey, + "Votifier throttling " + blockedKey + " (tunnel=" + tunnelMode + "), retry in " + + Math.max(0, retry / 1000) + "s"); + return null; + } + String challenge = receiver.getChallenge(); sendHandshakeIfNeeded(in, writer, challenge); diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 1a5c2b6..43d5243 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -59,6 +59,41 @@ public boolean isBlocked(String key) { return isBlocked(key, null); } + /** + * Checks only the aggregate identity, without consulting a per-client state. + * This is used before proxy headers are read so an already-blocked remote can + * be rejected without making a primary tunnel state apply to every proxied + * client identity. + */ + public boolean isAggregateBlocked(String aggregateKey) { + if (config == null || !config.enabled || aggregateKey == null) { + return false; + } + + synchronized (throttleStateLock) { + return isBlocked(getAggregateState(aggregateKey)); + } + } + + public long aggregateRetryAfterMs(String aggregateKey) { + if (config == null || !config.enabled || aggregateKey == null) { + return 0L; + } + + synchronized (throttleStateLock) { + return retryAfterMs(getAggregateState(aggregateKey)); + } + } + + public String aggregateBlockedKey(String aggregateKey) { + if (aggregateKey == null) return null; + synchronized (throttleStateLock) { + if (aggregateStates.containsKey(aggregateKey)) return aggregateKey; + return "aggregate-overflow:" + + Math.floorMod(aggregateKey.hashCode(), AGGREGATE_OVERFLOW_BUCKETS); + } + } + public boolean isBlocked(String key, String aggregateKey) { if (config == null || !config.enabled) { return false; diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java index aac78ef..cef1eae 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteConnectionHandlerTest.java @@ -10,6 +10,7 @@ import java.io.OutputStream; import java.net.ServerSocket; import java.net.Socket; +import java.net.SocketTimeoutException; import java.nio.charset.StandardCharsets; import java.security.Key; import java.security.KeyPair; @@ -280,6 +281,36 @@ public void testProxyCanProvideClientIdentityBeforeThrottleDecision() throws Exc } } + @Test + public void testAggregateBlockRejectsBeforeHandshakeAndPayloadWait() throws Exception { + receiver.setUseTokens(false); + ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("127.0.0.1"), "5s", 1, "10s", 1, + "10s", false, 999, "1s", "60s"); + VoteThrottleService throttleService = new VoteThrottleService(config); + for (int i = 0; i < 4096; i++) { + throttleService.fail("ip:filler:" + i, false, true); + } + throttleService.fail("ip:blocked", "tunnel:127.0.0.1", false, true); + assertTrue(throttleService.isAggregateBlocked("tunnel:127.0.0.1")); + + VoteConnectionHandler handler = new VoteConnectionHandler(receiver, throttleService); + try (ServerSocket serverSocket = new ServerSocket(0); + Socket client = new Socket("127.0.0.1", serverSocket.getLocalPort()); + Socket accepted = serverSocket.accept()) { + client.setSoTimeout(500); + Future future = executor.submit(() -> handler.handle(accepted)); + BufferedReader clientReader = new BufferedReader( + new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); + + try { + assertNull(clientReader.readLine(), "an aggregate-blocked remote must not receive a handshake"); + } catch (SocketTimeoutException ex) { + throw new AssertionError("aggregate rejection must happen before waiting for payload", ex); + } + assertNull(future.get(1, java.util.concurrent.TimeUnit.SECONDS)); + } + } + @Test public void testHandlePresentV1PayloadSkipsHandshake() throws Exception { receiver.setUseTokens(false); diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index ecfa977..5be9f01 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -508,6 +508,9 @@ public void testOverflowBucketIsIgnoredAfterAggregateCapacityRecovers() throws E assertEquals("aggregate-overflow:" + Math.floorMod("BB".hashCode(), 64), service.blockedKey("ip:overflow-b", "BB"), "all remotes rejected by one overflow bucket must share its log key"); + assertEquals("aggregate-overflow:" + Math.floorMod("BB".hashCode(), 64), + service.aggregateBlockedKey("BB"), + "pre-handshake aggregate rejection must use the same bounded log key"); Map aggregates = stateMap(service, "aggregateStates"); aggregates.remove(aggregates.keySet().iterator().next()); From ff3c94ca352292dfa9410d79772d21e553665efe Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 03:00:59 -0600 Subject: [PATCH 29/33] Bound saturated throttle maintenance --- .../votifier/net/VoteThrottleService.java | 31 ++++++++++++- .../tests/VoteReceiverThrottleTest.java | 44 +++++++++++++++++++ 2 files changed, 73 insertions(+), 2 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 43d5243..d5e886b 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -33,6 +33,8 @@ private static final class ThrottleState { private final ThrottleState[] aggregateOverflowStates = new ThrottleState[AGGREGATE_OVERFLOW_BUCKETS]; private final Object logStateLock = new Object(); private final Object throttleStateLock = new Object(); + /* Guarded by logStateLock; avoid O(n) sweeps while all entries are active. */ + private long nextLogSweepMs; /* Guarded by throttleStateLock; avoid O(n) sweeps while all entries are known active. */ private long nextThrottleSweepMs; private long nextAggregateSweepMs; @@ -322,6 +324,13 @@ private ThrottleState getAggregateState(String key) { if (state != null) { return state; } + long now = System.currentTimeMillis(); + if (aggregateStates.size() >= MAX_TRACKED_KEYS && trimAggregateStates(now)) { + state = aggregateStates.get(key); + if (state != null) { + return state; + } + } return aggregateStates.size() >= MAX_TRACKED_KEYS ? aggregateOverflowStates[Math.floorMod(key.hashCode(), AGGREGATE_OVERFLOW_BUCKETS)] : null; } @@ -339,15 +348,33 @@ private ThrottleState getOverflowAggregateState(String key, long now) { private void trimLogStates(long now) { if (logStates.size() < MAX_TRACKED_KEYS) { + nextLogSweepMs = 0L; return; } long expiry = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; + if (now < nextLogSweepMs) { + /* The map is still saturated, so make bounded progress without rescanning it. */ + removeOneIfFull(logStates); + return; + } + + long nextSweep = Long.MAX_VALUE; for (java.util.Map.Entry entry : logStates.entrySet()) { - if (now - entry.getValue().lastLogMs >= expiry) { + LogState state = entry.getValue(); + long stateExpiry = state.lastLogMs > Long.MAX_VALUE - expiry ? Long.MAX_VALUE + : state.lastLogMs + expiry; + if (stateExpiry <= now) { logStates.remove(entry.getKey(), entry.getValue()); + } else { + nextSweep = Math.min(nextSweep, stateExpiry); } } - removeOneIfFull(logStates); + if (logStates.size() >= MAX_TRACKED_KEYS) { + removeOneIfFull(logStates); + nextLogSweepMs = nextSweep; + } else { + nextLogSweepMs = 0L; + } } private boolean trimThrottleStates(long now) { diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 5be9f01..24f3a5a 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -321,6 +321,24 @@ public void testExistingLogKeyDoesNotEvictAnotherStateAtCapacity() throws Except assertEquals(4096, states.size()); } + @Test + public void testSaturatedLogStateCachesExpiryAndEvictsWithoutRescanning() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.allowLog("log:saturated:" + i, "message"); + } + + service.allowLog("log:saturated:first-miss", "message"); + long nextSweep = longField(service, "nextLogSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + service.allowLog("log:saturated:second-miss", "message"); + + assertEquals(nextSweep, longField(service, "nextLogSweepMs"), + "saturated log misses before expiry must reuse the cached sweep deadline"); + assertEquals(4096, mapSize(service, "logStates")); + } + @Test public void testOverflowDoesNotEvictActiveBan() throws Exception { VoteThrottleService service = new VoteThrottleService( @@ -571,6 +589,32 @@ public void testSaturatedPrimaryStateCachesItsNextPossibleSweep() throws Excepti "a miss before the earliest expiry must reuse the saturated-state decision"); } + @Test + public void testAggregateLookupReclaimsExpiredEntryBeforeUsingOverflowBucket() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:primary:" + i, false, false); + } + for (int i = 0; i < 4096; i++) { + service.fail("ip:aggregate:" + i, "remote:" + i, false, false); + } + + Map aggregates = stateMap(service, "aggregateStates"); + Object expired = aggregates.get("remote:0"); + assertNotNull(expired); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + windowStartField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextAggregateSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + assertFalse(service.isAggregateBlocked("remote:not-yet-tracked")); + assertEquals(4095, aggregates.size(), + "an expired aggregate must be reclaimed before an overflow bucket is consulted"); + } + private static ThrottleConfig tunnelCfg(String... remoteIps) { return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, "10s", 1, "10s", true, 1, "60s", "60s"); From 5d59a023df4fec00fb72dac74d303b6d6d2454f5 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 03:20:31 -0600 Subject: [PATCH 30/33] Retain saturated log expiry deadline --- .../votifier/net/VoteThrottleService.java | 9 ++++--- .../tests/VoteReceiverThrottleTest.java | 27 +++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index d5e886b..35af76a 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -348,7 +348,9 @@ private ThrottleState getOverflowAggregateState(String key, long now) { private void trimLogStates(long now) { if (logStates.size() < MAX_TRACKED_KEYS) { - nextLogSweepMs = 0L; + /* A miss immediately adds one entry. Preserve the cached deadline when + * that insertion will refill the map. */ + if (logStates.size() < MAX_TRACKED_KEYS - 1) nextLogSweepMs = 0L; return; } long expiry = config != null ? Math.max(250L, config.logWindowMs) : 60_000L; @@ -371,10 +373,9 @@ private void trimLogStates(long now) { } if (logStates.size() >= MAX_TRACKED_KEYS) { removeOneIfFull(logStates); - nextLogSweepMs = nextSweep; - } else { - nextLogSweepMs = 0L; } + /* Even when reclamation made one slot, allowLog immediately refills it. */ + nextLogSweepMs = nextSweep == Long.MAX_VALUE ? 0L : nextSweep; } private boolean trimThrottleStates(long now) { diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 24f3a5a..81e65d9 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -339,6 +339,33 @@ public void testSaturatedLogStateCachesExpiryAndEvictsWithoutRescanning() throws assertEquals(4096, mapSize(service, "logStates")); } + @Test + public void testReclaimedLogSlotRetainsNextExpiryAfterImmediateRefill() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 2, "10s", 2, "10s", false, 999, "1s")); + for (int i = 0; i < 4096; i++) { + service.allowLog("log:reclaim:" + i, "message"); + } + + Map logs = stateMap(service, "logStates"); + Object expired = logs.get("log:reclaim:0"); + Field lastLogField = expired.getClass().getDeclaredField("lastLogMs"); + lastLogField.setAccessible(true); + lastLogField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextLogSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + service.allowLog("log:reclaim:first-miss", "message"); + long nextSweep = longField(service, "nextLogSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + service.allowLog("log:reclaim:second-miss", "message"); + + assertEquals(nextSweep, longField(service, "nextLogSweepMs"), + "the refill after reclamation must retain the next active expiry"); + assertEquals(4096, logs.size()); + } + @Test public void testOverflowDoesNotEvictActiveBan() throws Exception { VoteThrottleService service = new VoteThrottleService( From b63ef9ab488d43b7a8826b71e1f6db5db3e43be2 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 03:49:30 -0600 Subject: [PATCH 31/33] Retain reclaimed throttle deadlines --- .../votifier/net/VoteThrottleService.java | 21 ++++- .../tests/VoteReceiverThrottleTest.java | 78 +++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 35af76a..7854914 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -44,6 +44,7 @@ public VoteThrottleService(ThrottleConfig config) { this.config = config; if (config != null) { for (String remoteIp : config.tunnelRemoteIps) { + if (aggregateStates.size() >= MAX_TRACKED_KEYS) break; aggregateStates.put("tunnel:" + remoteIp, new ThrottleState()); } } @@ -300,6 +301,9 @@ private ThrottleState getThrottleState(String key) { created.windowStartMs = System.currentTimeMillis(); ThrottleState existing = throttleStates.putIfAbsent(key, created); state = existing == null ? created : existing; + if (existing == null && throttleStates.size() >= MAX_TRACKED_KEYS) { + nextThrottleSweepMs = earlierDeadline(nextThrottleSweepMs, stateExpiry(created)); + } } return state; } @@ -316,9 +320,16 @@ private ThrottleState getAggregateThrottleState(String key, long now) { ThrottleState created = new ThrottleState(); created.windowStartMs = now; ThrottleState existing = aggregateStates.putIfAbsent(key, created); + if (existing == null && aggregateStates.size() >= MAX_TRACKED_KEYS) { + nextAggregateSweepMs = earlierDeadline(nextAggregateSweepMs, stateExpiry(created)); + } return existing == null ? created : existing; } + private static long earlierDeadline(long current, long candidate) { + return current == 0L ? candidate : Math.min(current, candidate); + } + private ThrottleState getAggregateState(String key) { ThrottleState state = aggregateStates.get(key); if (state != null) { @@ -392,7 +403,9 @@ private boolean trimThrottleStates(long now) { } else nextSweep = Math.min(nextSweep, stateExpiry(state)); } if (throttleStates.size() < MAX_TRACKED_KEYS) { - nextThrottleSweepMs = 0L; + /* Keep the earliest active expiry: the caller immediately refills the + * reclaimed slot, and the map can become saturated again before it. */ + nextThrottleSweepMs = nextSweep == Long.MAX_VALUE ? 0L : nextSweep; return true; } nextThrottleSweepMs = nextSweep; @@ -416,7 +429,11 @@ private boolean trimAggregateStates(long now) { } else nextSweep = Math.min(nextSweep, stateExpiry(state)); } boolean available = aggregateStates.size() < MAX_TRACKED_KEYS; - nextAggregateSweepMs = available ? 0L : nextSweep; + /* As with primary states, preserve the deadline across the immediate + * insertion that consumes a reclaimed aggregate slot. */ + nextAggregateSweepMs = nextSweep == Long.MAX_VALUE + ? available ? 0L : Long.MAX_VALUE + : nextSweep; return available; } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 81e65d9..2df6f91 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -616,6 +616,35 @@ public void testSaturatedPrimaryStateCachesItsNextPossibleSweep() throws Excepti "a miss before the earliest expiry must reuse the saturated-state decision"); } + @Test + public void testReclaimedPrimarySlotRetainsNextExpiryAfterImmediateRefill() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 999, "10s", 999, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:reclaim:" + index, false, true); + + Map states = stateMap(service, "throttleStates"); + Object expired = states.get("ip:reclaim:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long laterExistingExpiry = System.currentTimeMillis() + 60_000L; + for (Object state : states.values()) windowStartField.setLong(state, laterExistingExpiry); + windowStartField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextThrottleSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + service.fail("ip:reclaim:first-miss", false, true); + long nextSweep = longField(service, "nextThrottleSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + assertTrue(nextSweep < laterExistingExpiry, + "the newly inserted state's earlier expiry must bound the retained deadline"); + service.fail("ip:reclaim:second-miss", false, true); + + assertEquals(nextSweep, longField(service, "nextThrottleSweepMs"), + "the refill after reclamation must retain the next active expiry"); + assertEquals(4096, states.size()); + } + @Test public void testAggregateLookupReclaimsExpiredEntryBeforeUsingOverflowBucket() throws Exception { VoteThrottleService service = new VoteThrottleService( @@ -642,6 +671,55 @@ public void testAggregateLookupReclaimsExpiredEntryBeforeUsingOverflowBucket() t "an expired aggregate must be reclaimed before an overflow bucket is consulted"); } + @Test + public void testReclaimedAggregateSlotRetainsNextExpiryAfterImmediateRefill() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("5s", 999, "10s", 999, "10s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) { + service.fail("ip:aggregate-primary-fill:" + index, false, true); + } + for (int index = 0; index < 4096; index++) { + service.fail("ip:aggregate-reclaim:" + index, "remote:aggregate-reclaim:" + index, false, false); + } + + Map aggregates = stateMap(service, "aggregateStates"); + Object expired = aggregates.get("remote:aggregate-reclaim:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long laterExistingExpiry = System.currentTimeMillis() + 60_000L; + for (Object state : aggregates.values()) windowStartField.setLong(state, laterExistingExpiry); + windowStartField.setLong(expired, System.currentTimeMillis() - 6000L); + Field nextSweepField = VoteThrottleService.class.getDeclaredField("nextAggregateSweepMs"); + nextSweepField.setAccessible(true); + nextSweepField.setLong(service, 0L); + + service.fail("ip:aggregate-reclaim:first-miss", "remote:aggregate-reclaim:first-miss", false, false); + long nextSweep = longField(service, "nextAggregateSweepMs"); + assertTrue(nextSweep > System.currentTimeMillis()); + assertTrue(nextSweep < laterExistingExpiry, + "the newly inserted aggregate's earlier expiry must bound the retained deadline"); + service.fail("ip:aggregate-reclaim:second-miss", "remote:aggregate-reclaim:second-miss", false, false); + + assertEquals(nextSweep, longField(service, "nextAggregateSweepMs"), + "the aggregate refill after reclamation must retain the next active expiry"); + assertEquals(4096, aggregates.size()); + } + + @Test + public void testConfiguredAggregateIdentitiesRemainBoundedAndCacheSaturation() throws Exception { + java.util.Set configuredRemotes = new java.util.HashSet(); + for (int index = 0; index < 5000; index++) configuredRemotes.add("192.0.2." + index); + VoteThrottleService service = new VoteThrottleService( + new ThrottleConfig(true, configuredRemotes, "5s", 999, "10s", 999, "10s", false, 999, "1s", "1s")); + + assertEquals(4096, mapSize(service, "aggregateStates")); + assertFalse(service.isAggregateBlocked("remote:unknown")); + assertEquals(Long.MAX_VALUE, longField(service, "nextAggregateSweepMs"), + "a map containing only pinned configured identities must not be rescanned on every miss"); + assertFalse(service.isAggregateBlocked("remote:another-unknown")); + assertEquals(4096, mapSize(service, "aggregateStates")); + } + private static ThrottleConfig tunnelCfg(String... remoteIps) { return new ThrottleConfig(true, new java.util.HashSet(java.util.Arrays.asList(remoteIps)), "5s", 1, "10s", 1, "10s", true, 1, "60s", "60s"); From 9aa5068d7db48db03973c89dda2c5ad6d8ebbadf Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 04:08:43 -0600 Subject: [PATCH 32/33] Preserve shared aggregate throttle failures --- .../votifier/net/VoteThrottleService.java | 8 +++++++- .../tests/VoteReceiverThrottleTest.java | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 7854914..1bcf375 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -23,6 +23,9 @@ private static final class ThrottleState { private volatile int failures; private volatile long throttledUntilMs; private volatile long bannedUntilMs; + /* Once a proxied identity contributes to this aggregate, direct successes + * must not treat the aggregate as private to the tunnel remote. */ + private volatile boolean sharedByProxiedIdentity; } private final ThrottleConfig config; @@ -189,6 +192,9 @@ public void fail(String key, String aggregateKey, boolean tunnelMode, boolean re if (state == null) { return; } + if (aggregate && aggregateKey != null && !key.equals(aggregateKey)) { + state.sharedByProxiedIdentity = true; + } if (now - state.windowStartMs > config.windowMs) { state.windowStartMs = now; @@ -239,7 +245,7 @@ public void success(String key, String aggregateKey) { // bucket is deliberately shared by many identities and must not be // reset by one successful request. state = aggregateStates.get(aggregateKey); - if (state != null) { + if (state != null && !state.sharedByProxiedIdentity) { state.failures = 0; state.windowStartMs = now; if (!isConfiguredAggregateKey(aggregateKey) && state.bannedUntilMs <= now diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index 2df6f91..ff304c0 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -461,6 +461,23 @@ public void testProxiedSuccessDoesNotClearSharedAggregateFailures() { "a proxied success must not erase another identity's aggregate failure"); } + @Test + public void testDirectSuccessDoesNotClearAggregateSharedByProxiedFailures() { + VoteThrottleService service = new VoteThrottleService(tunnelCfg("proxy")); + for (int i = 0; i < 4096; i++) { + service.fail("ip:filler:" + i, false, true); + } + + String aggregateKey = "tunnel:proxy"; + service.fail("ip:proxied", aggregateKey, false, false); + assertTrue(service.isAggregateBlocked(aggregateKey)); + + service.success(aggregateKey, aggregateKey); + + assertTrue(service.isAggregateBlocked(aggregateKey), + "a direct success must not clear an aggregate that has been shared by proxied identities"); + } + @Test public void testFullMapPreservesInWindowCountersAndUsesAggregateTunnel() throws Exception { ThrottleConfig config = new ThrottleConfig(true, Collections.singleton("proxy"), "5s", 2, "10s", 2, From 8e531fb48105694f38fbe0d3f43c531bfbcc8e17 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Tue, 8 Sep 2026 04:36:47 -0600 Subject: [PATCH 33/33] Track earliest deadline at throttle capacity --- .../votifier/net/VoteThrottleService.java | 19 ++++++- .../tests/VoteReceiverThrottleTest.java | 49 +++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java index 1bcf375..126e1c5 100644 --- a/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java +++ b/VotifierPlus/src/main/java/com/vexsoftware/votifier/net/VoteThrottleService.java @@ -308,7 +308,9 @@ private ThrottleState getThrottleState(String key) { ThrottleState existing = throttleStates.putIfAbsent(key, created); state = existing == null ? created : existing; if (existing == null && throttleStates.size() >= MAX_TRACKED_KEYS) { - nextThrottleSweepMs = earlierDeadline(nextThrottleSweepMs, stateExpiry(created)); + nextThrottleSweepMs = nextThrottleSweepMs == 0L + ? earliestThrottleExpiry(throttleStates, false) + : earlierDeadline(nextThrottleSweepMs, stateExpiry(created)); } } return state; @@ -327,11 +329,24 @@ private ThrottleState getAggregateThrottleState(String key, long now) { created.windowStartMs = now; ThrottleState existing = aggregateStates.putIfAbsent(key, created); if (existing == null && aggregateStates.size() >= MAX_TRACKED_KEYS) { - nextAggregateSweepMs = earlierDeadline(nextAggregateSweepMs, stateExpiry(created)); + nextAggregateSweepMs = nextAggregateSweepMs == 0L + ? earliestThrottleExpiry(aggregateStates, true) + : earlierDeadline(nextAggregateSweepMs, stateExpiry(created)); } return existing == null ? created : existing; } + private long earliestThrottleExpiry(ConcurrentHashMap states, + boolean skipConfiguredAggregates) { + long earliest = Long.MAX_VALUE; + for (java.util.Map.Entry entry : states.entrySet()) { + if (!skipConfiguredAggregates || !isConfiguredAggregateKey(entry.getKey())) { + earliest = Math.min(earliest, stateExpiry(entry.getValue())); + } + } + return earliest; + } + private static long earlierDeadline(long current, long candidate) { return current == 0L ? candidate : Math.min(current, candidate); } diff --git a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java index ff304c0..249bc70 100644 --- a/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java +++ b/VotifierPlus/src/test/java/com/bencodez/votifierplus/tests/VoteReceiverThrottleTest.java @@ -633,6 +633,29 @@ public void testSaturatedPrimaryStateCachesItsNextPossibleSweep() throws Excepti "a miss before the earliest expiry must reuse the saturated-state decision"); } + @Test + public void testInitialPrimarySaturationUsesAnOlderStateExpiry() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("60s", 999, "60s", 999, "60s", false, 999, "1s")); + for (int index = 0; index < 4095; index++) service.fail("ip:initial-primary:" + index, false, true); + + Map states = stateMap(service, "throttleStates"); + Object expired = states.get("ip:initial-primary:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long now = System.currentTimeMillis(); + for (Object state : states.values()) windowStartField.setLong(state, now + 60_000L); + windowStartField.setLong(expired, now - 61_000L); + + service.fail("ip:initial-primary:capacity", false, true); + assertTrue(longField(service, "nextThrottleSweepMs") <= System.currentTimeMillis(), + "the first saturated map must retain the earlier existing expiry"); + service.fail("ip:initial-primary:replacement", false, true); + + assertFalse(states.containsKey("ip:initial-primary:0")); + assertTrue(states.containsKey("ip:initial-primary:replacement")); + } + @Test public void testReclaimedPrimarySlotRetainsNextExpiryAfterImmediateRefill() throws Exception { VoteThrottleService service = new VoteThrottleService( @@ -722,6 +745,32 @@ public void testReclaimedAggregateSlotRetainsNextExpiryAfterImmediateRefill() th assertEquals(4096, aggregates.size()); } + @Test + public void testInitialAggregateSaturationUsesAnOlderStateExpiry() throws Exception { + VoteThrottleService service = new VoteThrottleService( + cfg("60s", 999, "60s", 999, "60s", false, 999, "1s")); + for (int index = 0; index < 4096; index++) service.fail("ip:initial-aggregate-primary:" + index, false, true); + for (int index = 0; index < 4095; index++) { + service.fail("ip:initial-aggregate-client:" + index, "remote:initial-aggregate:" + index, false, false); + } + + Map aggregates = stateMap(service, "aggregateStates"); + Object expired = aggregates.get("remote:initial-aggregate:0"); + Field windowStartField = expired.getClass().getDeclaredField("windowStartMs"); + windowStartField.setAccessible(true); + long now = System.currentTimeMillis(); + for (Object state : aggregates.values()) windowStartField.setLong(state, now + 60_000L); + windowStartField.setLong(expired, now - 61_000L); + + service.fail("ip:initial-aggregate-client:capacity", "remote:initial-aggregate:capacity", false, false); + assertTrue(longField(service, "nextAggregateSweepMs") <= System.currentTimeMillis(), + "the first saturated aggregate map must retain the earlier existing expiry"); + service.fail("ip:initial-aggregate-client:replacement", "remote:initial-aggregate:replacement", false, false); + + assertFalse(aggregates.containsKey("remote:initial-aggregate:0")); + assertTrue(aggregates.containsKey("remote:initial-aggregate:replacement")); + } + @Test public void testConfiguredAggregateIdentitiesRemainBoundedAndCacheSaturation() throws Exception { java.util.Set configuredRemotes = new java.util.HashSet();