diff --git a/.mex/events/decisions.jsonl b/.mex/events/decisions.jsonl index 16851696d..a4406b6e4 100644 --- a/.mex/events/decisions.jsonl +++ b/.mex/events/decisions.jsonl @@ -1 +1,7 @@ +{"timestamp":"2026-09-21T10:51:59.003Z","kind":"decision","message":"Proxy-to-backend reward votes use a capability-negotiated durable outbox and completion acknowledgement. The guarantee is at least once; HTTP retains its existing delivery path, old backends retain legacy semantics, and exactly-once reward effects remain out of scope.","files":["VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutbox.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java"],"cwd":".","source":"agent","status":"implemented"} +{"timestamp":"2026-09-21T11:07:28.400Z","kind":"decision","message":"Reliable backend votes persist completed vote IDs for seven days before acknowledgement. Receipt persistence failure retries without repeating effects in the active process; restart deduplication covers durable completions, while a crash during non-transactional reward effects remains at-least-once.","files":["VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouter.java"],"cwd":".","source":"agent","status":"implemented"} +{"timestamp":"2026-09-21T11:41:29.374Z","kind":"decision","message":"Supersedes the seven-day receipt decision: completion receipts have no time expiry and fail closed at the bounded journal capacity, HTTP participates in the completion-acknowledged outbox, and receipt-write failures keep an in-memory post-effect fence until persistence succeeds.","files":["VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginProxy.java"],"cwd":".","source":"agent","status":"implemented"} +{"timestamp":"2026-09-21T12:13:09.371Z","kind":"decision","message":"Supersedes indefinite unreclaimed completion receipts: reliable vote delivery uses a durable two-phase receipt-release handshake. After the proxy durably records backend completion, it retries a release marker until the backend durably removes the receipt and acknowledges retirement; only then does the proxy remove the marker.","files":["VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutbox.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginWire.java"],"cwd":".","source":"agent","status":"implemented"} +{"timestamp":"2026-09-21T12:30:33.599Z","kind":"decision","message":"Supersedes immediate receipt removal after release: backend receipt release creates a 24-hour durable tombstone to fence deliveries already in flight, then reclaims it. Legacy downgrade delivery preserves the proxy release marker so a later capable backend can retire any receipt whose completion acknowledgement was lost.","files":["VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginProxy.java"],"cwd":".","source":"agent","status":"implemented"} +{"timestamp":"2026-09-21T12:46:34.284Z","kind":"decision","message":"Receipt releases persist a 24-hour tombstone even when the backend has no active receipt, fencing a late legacy retry. The bounded receipt store keeps completion headroom larger than the entire ordered lane plus separate tombstone capacity. Releases backed by an existing durable receipt may bypass a capacity-blocked vote because durable completion makes that reordering safe; unknown releases remain ordered.","files":["VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java","VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java"],"cwd":".","source":"agent","status":"implemented"} {"timestamp":"2026-09-21T10:57:35.704Z","kind":"decision","message":"Keep the common Linux x86_64 SQLite native in the downloadable JAR; provision other SQLite targets from the pinned, SHA-256-verified sqlite-jdbc artifact with a documented offline pre-provisioning path. Use SimpleAPI JDK-only TLS identity without Bouncy Castle, enforce a 10 MiB package gate, and alert on meaningful size growth.","files":["VotingPlugin/pom.xml","VotingPlugin/src/main/java/com/bencodez/votingplugin/util/SqliteNativeLibrary.java","docs/jar-packaging.md"],"cwd":".","source":"agent","status":"implemented"} diff --git a/AGENTS.md b/AGENTS.md index ddf7a3b93..307238465 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -186,6 +186,11 @@ credentials, generated JARs, dependency caches, IDE output, or unrelated formatt - Trace whether the code runs on the connector worker, proxy thread, Bukkit primary thread, or a SQL executor. - Preserve queued votes across saturation, shutdown, and restart; overflow handling must be bounded, durable when promised, and observable rather than silently dropping work. +- Proxy-to-backend guaranteed delivery is capability negotiated and at least once. Journal a reward-bearing envelope before + reporting transport acceptance, retain it until the matching backend completion acknowledgement is durable, persist + completed IDs before acknowledgement for restart-safe deduplication, retire receipts only through the durable + proxy-confirmed release handshake, retain a bounded durable tombstone for in-flight retries, and keep legacy send + behavior for backends that do not advertise the capability. - Treat scheduler units explicitly. Verify whether each delay is in ticks, milliseconds, or seconds, especially across Bukkit, Folia, BungeeCord, and Velocity adapters. - Register listeners and lifecycle wakeups before producers can publish work; startup/reload ordering must not strand already-persisted or newly-arriving operations. - Protocol-mode changes must not silently broaden legacy v1/RSA acceptance when token-only operation is configured or intended; cover downgrade behavior with tests. diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java index da451cb49..d554a2ee2 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java @@ -175,7 +175,7 @@ public File getLoadedPluginJarFile() { @Getter private BackendProxyHandler backendProxyHandler; - private final ProcessedVoteCache backendProcessedVoteCache = new ProcessedVoteCache(); + private ProcessedVoteCache backendProcessedVoteCache; private BackendOrderedVoteOverflowQueue backendOrderedVoteOverflowQueue; private final AtomicReference backendPluginMessageTarget = new AtomicReference<>(); private PluginMessageHandler backendPluginMessageRelay; @@ -558,8 +558,16 @@ private synchronized BackendOrderedVoteOverflowQueue getOrCreateBackendOrderedVo return backendOrderedVoteOverflowQueue; } + private synchronized ProcessedVoteCache getOrCreateBackendProcessedVoteCache() { + if (backendProcessedVoteCache == null) { + backendProcessedVoteCache = new ProcessedVoteCache( + new File(getDataFolder(), "BackendProcessedVotes.dat").toPath()); + } + return backendProcessedVoteCache; + } + private void loadBungeeHandler() { - BackendProxyHandler candidate = new BackendProxyHandler(this, backendProcessedVoteCache, + BackendProxyHandler candidate = new BackendProxyHandler(this, getOrCreateBackendProcessedVoteCache(), getOrCreateBackendOrderedVoteOverflowQueue()); try { candidate.load(); @@ -1435,7 +1443,7 @@ public synchronized BackendProxyRestart prepareBackendProxyHandlerRestart() { boolean previousRequiresPreparation = previous != null && (deferredReplacementLoad || previous.requiresPreparationForReplacement() || previous.requiresRedisRetirement()); - BackendProxyHandler replacement = new BackendProxyHandler(this, backendProcessedVoteCache, + BackendProxyHandler replacement = new BackendProxyHandler(this, getOrCreateBackendProcessedVoteCache(), getOrCreateBackendOrderedVoteOverflowQueue()); if (!deferredReplacementLoad) { try { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendOrderedVoteOverflowQueue.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendOrderedVoteOverflowQueue.java index 73be55ef3..0cef7d6a3 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendOrderedVoteOverflowQueue.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendOrderedVoteOverflowQueue.java @@ -434,7 +434,8 @@ private static boolean isOrderedVoteMessage(JsonEnvelope envelope) { String subChannel = envelope.getSubChannel(); return VotingPluginWire.SUB_VOTE.equals(subChannel) || VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel) - || VotingPluginWire.SUB_VOTE_UPDATE.equals(subChannel); + || VotingPluginWire.SUB_VOTE_UPDATE.equals(subChannel) + || VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE.equals(subChannel); } private void requestPersistenceLocked() { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java index 744efcf06..5ddd1c0fc 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java @@ -56,6 +56,7 @@ public class BackendProxyHandler implements Listener { private boolean orderedVoteDispatchPaused = true; private boolean orderedVoteDispatchClosing; private boolean orderedVoteQuarantineFailed; + private final AtomicBoolean durableReceiptReleaseActive = new AtomicBoolean(); private JsonEnvelope orderedVoteDispatchInFlight; private BackendOrderedVoteOverflowQueue.PendingEnvelope orderedVoteOverflowInFlight; private JsonEnvelope orderedVoteShutdownQuarantined; @@ -229,10 +230,49 @@ private boolean isOrderedVoteMessage(JsonEnvelope envelope) { String subChannel = envelope.getSubChannel(); return VotingPluginWire.SUB_VOTE.equals(subChannel) || VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel) - || VotingPluginWire.SUB_VOTE_UPDATE.equals(subChannel); + || VotingPluginWire.SUB_VOTE_UPDATE.equals(subChannel) + || VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE.equals(subChannel); } private void dispatchOrderedVote(JsonEnvelope envelope, Runnable ignoredLocalDispatch) { + BackendProxyMessageRouter router = messageRouter; + if (dispatchDurableReceiptRelease(router, envelope)) return; + enqueueOrderedVote(envelope, ignoredLocalDispatch); + } + + private boolean dispatchDurableReceiptRelease(BackendProxyMessageRouter router, JsonEnvelope envelope) { + if (router == null || !router.hasDurableReceiptForRelease(envelope)) return false; + // The proxy durably retries unacknowledged releases. Keep every known release + // outside the ordered vote lane, including a release received while this + // bounded single-flight worker is busy. + if (!durableReceiptReleaseActive.compareAndSet(false, true)) return true; + try { + plugin.getBukkitScheduler().runTaskAsynchronously(plugin, + () -> processDurableReceiptRelease(router, envelope)); + } catch (RuntimeException schedulingFailure) { + plugin.debug(schedulingFailure); + processDurableReceiptRelease(router, envelope); + } + return true; + } + + private void processDurableReceiptRelease(BackendProxyMessageRouter router, JsonEnvelope envelope) { + AtomicBoolean completed = new AtomicBoolean(); + java.util.function.Consumer completion = outcome -> { + if (!completed.compareAndSet(false, true)) return; + durableReceiptReleaseActive.set(false); + }; + try { + router.handleOrderedVote(envelope, completion); + } catch (RuntimeException | Error failure) { + if (completed.compareAndSet(false, true)) { + durableReceiptReleaseActive.set(false); + } + throw failure; + } + } + + private void enqueueOrderedVote(JsonEnvelope envelope, Runnable ignoredLocalDispatch) { BackendProxyHandler handoffTarget; synchronized (orderedVoteDispatch) { handoffTarget = orderedVoteHandoffTarget; @@ -390,6 +430,12 @@ private void runNextOrderedVoteDispatch() { } }; try { + if (dispatchDurableReceiptRelease(messageRouter, next)) { + // This local copy may predate the dedicated release lane. The proxy retains + // the authoritative durable retry until it receives an acknowledgement. + complete.accept(OrderedVoteOutcome.COMPLETE); + return; + } messageRouter.handleOrderedVote(next, complete); } catch (RuntimeException | Error failure) { complete.accept(OrderedVoteOutcome.QUARANTINE); @@ -417,7 +463,7 @@ private void finishOrderedVoteDispatch(BackendOrderedVoteOverflowQueue.PendingEn boolean successful = outcome == OrderedVoteOutcome.COMPLETE; if (successful && overflowEntry != null && orderedVoteOverflow != null) { orderedVoteOverflow.acknowledgeAsync(overflowEntry, - stored -> completeOrderedVoteAcknowledgement(stored)); + stored -> completeOrderedVoteAcknowledgement(stored, envelope)); return; } synchronized (orderedVoteDispatch) { @@ -432,9 +478,10 @@ private void finishOrderedVoteDispatch(BackendOrderedVoteOverflowQueue.PendingEn if (successful) scheduleOrderedVoteDispatchLocked(); else retryOrderedVoteDispatchLocked(); } + if (successful) sendVoteDeliveryAcknowledgement(envelope); } - private void completeOrderedVoteAcknowledgement(boolean stored) { + private void completeOrderedVoteAcknowledgement(boolean stored, JsonEnvelope envelope) { synchronized (orderedVoteDispatch) { if (stored) { orderedVoteDispatchInFlight = null; @@ -450,6 +497,23 @@ private void completeOrderedVoteAcknowledgement(boolean stored) { orderedVoteDispatch.notifyAll(); if (stored) scheduleOrderedVoteDispatchLocked(); } + if (stored) sendVoteDeliveryAcknowledgement(envelope); + } + + private void sendVoteDeliveryAcknowledgement(JsonEnvelope envelope) { + if ((!VotingPluginWire.SUB_VOTE.equals(envelope.getSubChannel()) + && !VotingPluginWire.SUB_VOTE_ONLINE.equals(envelope.getSubChannel())) + || !VotingPluginWire.requestsVoteDeliveryAcknowledgement(envelope) + || globalMessageHandler == null) return; + String voteId = envelope.getFields().get(VotingPluginWire.K_VOTE_ID); + try { + UUID parsed = voteId == null || voteId.isBlank() ? null : UUID.fromString(voteId); + if (parsed == null) return; + globalMessageHandler.sendMessage(VotingPluginWire.voteDeliveryAcknowledgement( + plugin.getBungeeSettings().getServer(), parsed, envelope.getSubChannel())); + } catch (IllegalArgumentException invalidVoteId) { + plugin.debug("Unable to acknowledge proxy vote with invalid vote ID"); + } } private void completeOrderedVoteQuarantine(BackendOrderedVoteOverflowQueue.PendingEnvelope overflowEntry, diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java new file mode 100644 index 000000000..50551a6a2 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/DurableVoteReceiptStore.java @@ -0,0 +1,212 @@ +package com.bencodez.votingplugin.backendproxy.cache; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +import com.bencodez.votingplugin.util.DurableFiles; + +/** Bounded append journal for backend vote IDs completed before acknowledgement. */ +final class DurableVoteReceiptStore { + private static final int MAX_ACTIVE_RECEIPTS = 262144; + /* Larger than the complete in-memory and durable ordered lane (256 + 512). */ + private static final int COMPLETION_HEADROOM = 1024; + /* One proxy cannot retain more release markers than its bounded outbox. */ + private static final int MAX_RELEASE_TOMBSTONES = 4096; + private static final long MAX_FILE_BYTES = 16L * 1024L * 1024L; + private static final String HEADER = "VP-VOTE-RECEIPTS-1"; + private static final String RELEASE = "R"; + static final long RELEASE_TOMBSTONE_TTL_MILLIS = TimeUnit.HOURS.toMillis(24); + private static final ConcurrentHashMap FILE_LOCKS = new ConcurrentHashMap<>(); + + private final Path file; + private final Object fileLock; + private final int maxActiveReceipts; + private final int completionHeadroom; + private final int maxReleaseTombstones; + private final LinkedHashMap receipts = new LinkedHashMap<>(); + private int activeReceipts; + private int releaseTombstones; + private int journalRecords; + + DurableVoteReceiptStore(Path file) throws IOException { + this(file, MAX_ACTIVE_RECEIPTS, COMPLETION_HEADROOM, MAX_RELEASE_TOMBSTONES); + } + + DurableVoteReceiptStore(Path file, int maxActiveReceipts, int completionHeadroom, + int maxReleaseTombstones) throws IOException { + this.file = file.toAbsolutePath().normalize(); + this.fileLock = FILE_LOCKS.computeIfAbsent(this.file, ignored -> new Object()); + this.maxActiveReceipts = maxActiveReceipts; + this.completionHeadroom = completionHeadroom; + this.maxReleaseTombstones = maxReleaseTombstones; + synchronized (fileLock) { + load(); + } + } + + synchronized Map snapshot() { + cleanupReleasedTombstones(System.currentTimeMillis()); + return new LinkedHashMap<>(receipts); + } + + synchronized boolean contains(UUID voteId) { + cleanupReleasedTombstones(System.currentTimeMillis()); + return voteId != null && receipts.containsKey(voteId); + } + + synchronized long complete(UUID voteId) { + if (voteId == null) return 0L; + cleanupReleasedTombstones(System.currentTimeMillis()); + Long current = receipts.get(voteId); + if (current != null) return current; + if (activeReceipts >= maxActiveReceipts + completionHeadroom) return 0L; + long expiresAt = Long.MAX_VALUE; + String record = voteId + "\t" + expiresAt + '\n'; + synchronized (fileLock) { + if (!prepareAppend(record) || !append(record)) return 0L; + } + putReceipt(voteId, expiresAt); + journalRecords++; + return expiresAt; + } + + synchronized long release(UUID voteId) { + if (voteId == null) return 0L; + long now = System.currentTimeMillis(); + cleanupReleasedTombstones(now); + Long current = receipts.get(voteId); + if (current != null && current != Long.MAX_VALUE) return current; + if (releaseTombstones >= maxReleaseTombstones) return 0L; + long expiresAt = now + RELEASE_TOMBSTONE_TTL_MILLIS; + String record = RELEASE + '\t' + voteId + '\t' + expiresAt + '\n'; + synchronized (fileLock) { + if (!prepareAppend(record) || !append(record)) return 0L; + } + putReceipt(voteId, expiresAt); + journalRecords++; + return expiresAt; + } + + private void load() throws IOException { + if (!Files.exists(file, LinkOption.NOFOLLOW_LINKS)) return; + if (Files.isSymbolicLink(file) || !Files.isRegularFile(file, LinkOption.NOFOLLOW_LINKS)) { + throw new IOException("Vote receipt journal is not a regular file"); + } + if (Files.size(file) > MAX_FILE_BYTES) throw new IOException("Vote receipt journal exceeds size limit"); + String content = Files.readString(file, StandardCharsets.UTF_8); + String[] lines = content.split("\\n", -1); + if (lines.length == 0 || !HEADER.equals(lines[0])) throw new IOException("Unsupported vote receipt journal"); + boolean unterminatedTail = !content.endsWith("\n"); + int completeLineLimit = unterminatedTail ? lines.length - 1 : lines.length; + for (int index = 1; index < completeLineLimit; index++) { + if (lines[index].isBlank()) continue; + try { + String[] fields = lines[index].split("\\t", 3); + if (RELEASE.equals(fields[0])) { + if (fields.length == 2) removeReceipt(UUID.fromString(fields[1])); + else if (fields.length == 3) { + UUID voteId = UUID.fromString(fields[1]); + long expiresAt = Long.parseLong(fields[2]); + if (expiresAt > System.currentTimeMillis()) putReceipt(voteId, expiresAt); + else removeReceipt(voteId); + } else throw new IllegalArgumentException("Malformed receipt release"); + } else { + if (fields.length != 2) throw new IllegalArgumentException("Malformed receipt"); + UUID voteId = UUID.fromString(fields[0]); + long expiresAt = Long.parseLong(fields[1]); + putReceipt(voteId, expiresAt); + } + } catch (RuntimeException malformed) { + throw new IOException("Malformed vote receipt journal", malformed); + } + journalRecords++; + if (activeReceipts > maxActiveReceipts + completionHeadroom + || releaseTombstones > maxReleaseTombstones) { + throw new IOException("Vote receipt journal exceeds entry limit"); + } + } + if (unterminatedTail && !compact()) throw new IOException("Unable to repair vote receipt journal"); + } + + private boolean prepareAppend(String record) { + try { + long projected = (Files.exists(file) ? Files.size(file) : HEADER.length() + 1L) + + record.getBytes(StandardCharsets.UTF_8).length; + boolean shouldCompact = journalRecords > receipts.size() * 2 + 256; + if ((projected > MAX_FILE_BYTES || shouldCompact) && compact()) { + projected = Files.size(file) + record.getBytes(StandardCharsets.UTF_8).length; + } + return projected <= MAX_FILE_BYTES; + } catch (IOException failure) { + return false; + } + } + + private boolean append(String record) { + try { + Files.createDirectories(file.getParent()); + if (!Files.exists(file, LinkOption.NOFOLLOW_LINKS)) { + Path staged = file.resolveSibling(file.getFileName() + ".tmp"); + Files.writeString(staged, HEADER + '\n' + record, StandardCharsets.UTF_8, + StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE); + DurableFiles.publishStagedFile(staged, file); + } else { + Files.writeString(file, record, StandardCharsets.UTF_8, StandardOpenOption.APPEND, + StandardOpenOption.WRITE, LinkOption.NOFOLLOW_LINKS); + DurableFiles.forceFile(file); + } + return true; + } catch (IOException failure) { + return false; + } + } + + private boolean compact() { + try { + StringBuilder content = new StringBuilder(HEADER).append('\n'); + for (Map.Entry receipt : receipts.entrySet()) { + content.append(receipt.getKey()).append('\t').append(receipt.getValue()).append('\n'); + } + Path staged = file.resolveSibling(file.getFileName() + ".tmp"); + Files.writeString(staged, content, StandardCharsets.UTF_8, StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE); + DurableFiles.publishStagedFile(staged, file); + journalRecords = receipts.size(); + return true; + } catch (IOException failure) { + return false; + } + } + + private void cleanupReleasedTombstones(long now) { + int before = receipts.size(); + receipts.entrySet().removeIf(entry -> entry.getValue() != Long.MAX_VALUE && entry.getValue() <= now); + releaseTombstones -= before - receipts.size(); + } + + private void putReceipt(UUID voteId, long expiresAt) { + Long previous = receipts.put(voteId, expiresAt); + if (previous != null) { + if (previous == Long.MAX_VALUE) activeReceipts--; + else releaseTombstones--; + } + if (expiresAt == Long.MAX_VALUE) activeReceipts++; + else releaseTombstones++; + } + + private void removeReceipt(UUID voteId) { + Long previous = receipts.remove(voteId); + if (previous == null) return; + if (previous == Long.MAX_VALUE) activeReceipts--; + else releaseTombstones--; + } +} diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java index 5029b7e4d..ccc8f8cdf 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java @@ -1,5 +1,7 @@ package com.bencodez.votingplugin.backendproxy.cache; +import java.io.IOException; +import java.nio.file.Path; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.Map; @@ -21,7 +23,10 @@ public class ProcessedVoteCache { @Getter private final ConcurrentHashMap processedVotes = new ConcurrentHashMap<>(); + private final java.util.Set completedVotes = ConcurrentHashMap.newKeySet(); + private final java.util.Set completedAwaitingReceipt = ConcurrentHashMap.newKeySet(); private final long ttlMillis; + private final DurableVoteReceiptStore durableReceipts; private final LinkedHashMap processedRedisDeliveries = new LinkedHashMap<>(); private final LinkedHashMap legacyRedisDeliveries = new LinkedHashMap<>(); private long legacyRedisDeliveryBytes; @@ -30,11 +35,37 @@ public class ProcessedVoteCache { private Object standbyRedisSubscriber; public ProcessedVoteCache() { - this(DEFAULT_TTL_MILLIS); + this(DEFAULT_TTL_MILLIS, (DurableVoteReceiptStore) null); } public ProcessedVoteCache(long ttlMillis) { + this(ttlMillis, (DurableVoteReceiptStore) null); + } + + public ProcessedVoteCache(Path receiptFile) { + this(DEFAULT_TTL_MILLIS, loadReceipts(receiptFile)); + } + + ProcessedVoteCache(long ttlMillis, Path receiptFile) { + this(ttlMillis, loadReceipts(receiptFile)); + } + + private ProcessedVoteCache(long ttlMillis, DurableVoteReceiptStore durableReceipts) { this.ttlMillis = ttlMillis; + this.durableReceipts = durableReceipts; + if (durableReceipts != null) { + Map receipts = durableReceipts.snapshot(); + processedVotes.putAll(receipts); + completedVotes.addAll(receipts.keySet()); + } + } + + private static DurableVoteReceiptStore loadReceipts(Path receiptFile) { + try { + return new DurableVoteReceiptStore(receiptFile); + } catch (IOException failure) { + throw new IllegalStateException("Unable to load durable backend vote receipts", failure); + } } public boolean reserve(UUID voteId) { @@ -46,6 +77,7 @@ public boolean reserve(UUID voteId) { long expiresAt = now + ttlMillis; while (true) { + if (completedAwaitingReceipt.contains(voteId)) return false; Long currentExpiry = processedVotes.get(voteId); if (currentExpiry == null) { if (processedVotes.putIfAbsent(voteId, expiresAt) == null) { @@ -60,12 +92,55 @@ public boolean reserve(UUID voteId) { } if (processedVotes.replace(voteId, currentExpiry, expiresAt)) { + completedVotes.remove(voteId); cleanup(now); return true; } } } + /** Persists successful processing before the backend emits a delivery acknowledgement. */ + public boolean complete(UUID voteId) { + if (voteId == null) return true; + completedAwaitingReceipt.add(voteId); + if (durableReceipts == null) { + completedVotes.add(voteId); + completedAwaitingReceipt.remove(voteId); + return true; + } + long expiresAt = durableReceipts.complete(voteId); + if (expiresAt <= 0L) return false; + processedVotes.put(voteId, expiresAt); + completedVotes.add(voteId); + completedAwaitingReceipt.remove(voteId); + return true; + } + + /** Returns whether vote effects completed, including a receipt append awaiting retry. */ + public boolean hasCompletedEffects(UUID voteId) { + return voteId != null && (completedVotes.contains(voteId) || completedAwaitingReceipt.contains(voteId)); + } + + /** Returns whether an acknowledgement-safe receipt is already durable. */ + public boolean hasDurableReceipt(UUID voteId) { + return durableReceipts != null && durableReceipts.contains(voteId); + } + + /** Durably retires a completed receipt after the proxy confirms outbox removal. */ + public boolean releaseCompletedReceipt(UUID voteId) { + if (voteId == null) return false; + if (durableReceipts == null) { + completedVotes.remove(voteId); + processedVotes.remove(voteId); + return true; + } + long expiresAt = durableReceipts.release(voteId); + if (expiresAt <= 0L) return false; + completedVotes.add(voteId); + processedVotes.put(voteId, expiresAt); + return true; + } + /** Deduplicates one Redis envelope across overlapping subscribers during a validated handoff. */ public synchronized boolean reserveRedisDelivery(String deliveryId) { if (deliveryId == null || !deliveryId.matches("[0-9a-fA-F-]{36}")) return true; @@ -167,7 +242,9 @@ public synchronized void unregisterRedisSubscriber(Object subscriber) { } private void cleanup(long now) { - processedVotes.entrySet().removeIf(entry -> entry.getValue() <= now); + processedVotes.forEach((voteId, expiresAt) -> { + if (expiresAt <= now && processedVotes.remove(voteId, expiresAt)) completedVotes.remove(voteId); + }); } /** Returns an exact UTF-8 length up to the per-delivery cap, then cap + 1. */ diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouter.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouter.java index a68238669..93dbd3594 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouter.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouter.java @@ -35,6 +35,7 @@ public enum OrderedVoteOutcome { COMPLETE, RETRY, QUARANTINE } private final BackendGlobalDataSync globalDataSync; private final BackendVotePartySync votePartySync; private final ProcessedVoteCache processedVoteCache; + private GlobalMessageHandler messages; public BackendProxyMessageRouter(VotingPluginMain plugin, BackendPresenceManager presenceManager, BackendGlobalDataSync globalDataSync, BackendVotePartySync votePartySync, @@ -47,6 +48,7 @@ public BackendProxyMessageRouter(VotingPluginMain plugin, BackendPresenceManager } public void register(GlobalMessageHandler messages, BungeeMethod method) { + this.messages = messages; messages.addListener(new GlobalMessageListener(VotingPluginWire.SUB_VOTE) { @Override public void onReceive(JsonEnvelope msg) { handleWireVote(msg); } }); @@ -56,6 +58,9 @@ public void register(GlobalMessageHandler messages, BungeeMethod method) { messages.addListener(new GlobalMessageListener(VotingPluginWire.SUB_VOTE_DELAY_REJECTED) { @Override public void onReceive(JsonEnvelope msg) { handleWireVoteDelayRejected(msg); } }); + messages.addListener(new GlobalMessageListener(VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE) { + @Override public void onReceive(JsonEnvelope msg) { handleVoteDeliveryReceiptRelease(messages, msg); } + }); messages.addListener(new GlobalMessageListener(VotingPluginWire.SUB_CONTROL_ENROLLMENT_RESULT) { @Override public void onReceive(JsonEnvelope msg) { plugin.handleBackendControlEnrollmentResult(msg); } }); @@ -82,6 +87,8 @@ public void register(GlobalMessageHandler messages, BungeeMethod method) { @Override public void onReceive(JsonEnvelope msg) { HashMap out = new HashMap<>(); out.put(VotingPluginWire.K_SERVER, nvl(plugin.getOptions().getServer())); + out.put(VotingPluginWire.K_VOTE_DELIVERY_ACK_VERSION, + VotingPluginWire.VOTE_DELIVERY_ACK_VERSION); String requestId = nvl(msg.getFields().get(VotingPluginWire.K_REQUEST_ID)); if (!requestId.isEmpty()) out.put(VotingPluginWire.K_REQUEST_ID, requestId); sendSubChannel(messages, VotingPluginWire.SUB_STATUS_OKAY, out); @@ -119,13 +126,28 @@ void handleVoteUpdate(JsonEnvelope msg) { public void handleOrderedVote(JsonEnvelope msg, Consumer completion) { if (completion == null) throw new IllegalArgumentException("Ordered vote completion is required"); String subChannel = msg.getSubChannel(); + if (VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE.equals(subChannel)) { + completion.accept(handleVoteDeliveryReceiptRelease(messages, msg)); + return; + } if (VotingPluginWire.SUB_VOTE_UPDATE.equals(subChannel)) { handleVoteUpdateWithOutcome(msg, completion); return; } if (VotingPluginWire.SUB_VOTE.equals(subChannel) || VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel)) { try { - handleWireVote(msg); + WireVoteResult result = handleWireVote(msg); + if (VotingPluginWire.requestsVoteDeliveryAcknowledgement(msg) + && (result == null || !result.effectsComplete())) { + completion.accept(OrderedVoteOutcome.QUARANTINE); + return; + } + UUID completedVoteId = result == null ? null : result.voteId(); + if (VotingPluginWire.requestsVoteDeliveryAcknowledgement(msg) + && completedVoteId != null && !processedVoteCache.complete(completedVoteId)) { + completion.accept(OrderedVoteOutcome.RETRY); + return; + } } catch (RuntimeException | Error failure) { completion.accept(OrderedVoteOutcome.QUARANTINE); throw failure; @@ -137,6 +159,12 @@ public void handleOrderedVote(JsonEnvelope msg, Consumer com throw new IllegalArgumentException("Unsupported ordered proxy vote message: " + subChannel); } + /** Allows a durable receipt release to bypass a capacity-blocked ordered vote. */ + public boolean hasDurableReceiptForRelease(JsonEnvelope msg) { + UUID voteId = validReceiptReleaseVoteId(msg); + return voteId != null && processedVoteCache.hasDurableReceipt(voteId); + } + /** * Processes one ordered VoteUpdate. User identity and shared cache population * are allowed to leave the platform thread, while offline reward/Bukkit work @@ -344,18 +372,18 @@ private void handleWireVoteDelayRejected(JsonEnvelope msg) { voteSite.giveWaitUntilVoteDelayRewards(user, rejected.wasOnline && user.isOnline(), true); } - private void handleWireVote(JsonEnvelope msg) { + private WireVoteResult handleWireVote(JsonEnvelope msg) { if (!validSchema(msg)) { - return; + return null; } VotingPluginWire.Vote vote = VotingPluginWire.readVote(msg); if (vote.uuid == null || vote.uuid.isEmpty()) { - return; + return null; } if (!ServiceSiteValidator.isValid(vote.service)) { plugin.getLogger().warning("Rejected proxy vote with invalid service site '" + ServiceSiteValidator.sanitizeForLog(vote.service) + "'"); - return; + return null; } plugin.debug("wire vote received from " + ServiceSiteValidator.sanitizeForLog(vote.player) + "/" @@ -368,7 +396,7 @@ private void handleWireVote(JsonEnvelope msg) { plugin.debug("Ignoring duplicate wire vote " + voteId + " for " + ServiceSiteValidator.sanitizeForLog(vote.player) + " on " + ServiceSiteValidator.sanitizeForLog(vote.service)); - return; + return new WireVoteResult(voteId, processedVoteCache.hasCompletedEffects(voteId)); } UUID javaUuid; @@ -377,7 +405,7 @@ private void handleWireVote(JsonEnvelope msg) { } catch (IllegalArgumentException e) { plugin.getLogger().warning("Invalid UUID in proxy vote: " + ServiceSiteValidator.sanitizeForLog(vote.uuid)); - return; + return new WireVoteResult(voteId, false); } VotingPluginUser user = plugin.getVotingPluginUserManager().getVotingPluginUser(javaUuid, vote.player); votePartySync.replace(totals.getVotePartyCurrent(), totals.getVotePartyRequired()); @@ -390,6 +418,10 @@ private void handleWireVote(JsonEnvelope msg) { if (vote.service != null && !vote.service.isEmpty()) { plugin.getServerData().addServiceSite(vote.service); } + return new WireVoteResult(voteId, true); + } + + private record WireVoteResult(UUID voteId, boolean effectsComplete) { } private boolean validSchema(JsonEnvelope msg) { @@ -401,6 +433,39 @@ private boolean validSchema(JsonEnvelope msg) { return false; } + private OrderedVoteOutcome handleVoteDeliveryReceiptRelease(GlobalMessageHandler messages, JsonEnvelope msg) { + if (messages == null) return OrderedVoteOutcome.QUARANTINE; + UUID voteId = validReceiptReleaseVoteId(msg); + if (voteId == null) return OrderedVoteOutcome.QUARANTINE; + String subChannel = nvl(msg.getFields().get(VotingPluginWire.K_VOTE_DELIVERY_SUBCHANNEL)); + try { + if (processedVoteCache.releaseCompletedReceipt(voteId)) { + messages.sendMessage(VotingPluginWire.voteDeliveryReceiptReleaseAcknowledgement( + plugin.getOptions().getServer(), voteId, subChannel)); + return OrderedVoteOutcome.COMPLETE; + } + return OrderedVoteOutcome.RETRY; + } catch (IllegalArgumentException invalidVoteId) { + plugin.debug("Ignored vote receipt release with invalid vote ID"); + return OrderedVoteOutcome.QUARANTINE; + } + } + + private UUID validReceiptReleaseVoteId(JsonEnvelope msg) { + if (msg == null || !VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE.equals(msg.getSubChannel()) + || !VotingPluginWire.requestsVoteDeliveryAcknowledgement(msg)) return null; + String server = nvl(msg.getFields().get(VotingPluginWire.K_SERVER)); + if (!plugin.getOptions().getServer().equalsIgnoreCase(server)) return null; + String subChannel = nvl(msg.getFields().get(VotingPluginWire.K_VOTE_DELIVERY_SUBCHANNEL)); + if (!VotingPluginWire.SUB_VOTE.equals(subChannel) + && !VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel)) return null; + try { + return UUID.fromString(nvl(msg.getFields().get(VotingPluginWire.K_VOTE_ID))); + } catch (IllegalArgumentException invalidVoteId) { + return null; + } + } + private void sendSubChannel(GlobalMessageHandler messages, String subChannel, HashMap fields) { JsonEnvelope.Builder builder = JsonEnvelope.builder(subChannel).schema(VotingPluginWire.SCHEMA_VERSION); for (Map.Entry entry : fields.entrySet()) { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutbox.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutbox.java new file mode 100644 index 000000000..13e372b7c --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutbox.java @@ -0,0 +1,227 @@ +package com.bencodez.votingplugin.proxy; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.Base64; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.UUID; + +import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope; +import com.bencodez.simpleapi.servercomm.codec.JsonEnvelopeCodec; +import com.bencodez.votingplugin.util.DurableFiles; + +/** Durable proxy outbox for reward-bearing votes awaiting backend completion. */ +final class ReliableVoteDeliveryOutbox { + static final int MAX_ENTRIES = 4096; + private static final long MAX_FILE_BYTES = 16L * 1024L * 1024L; + private static final String HEADER = "VP-VOTE-OUTBOX-2"; + private static final String ADD = "A"; + private static final String COMPLETED = "C"; + private static final String REMOVE = "R"; + + record Entry(String server, JsonEnvelope envelope, boolean awaitingReceiptRelease) { } + + private final Path file; + private final LinkedHashMap entries = new LinkedHashMap<>(); + private int journalRecords; + + ReliableVoteDeliveryOutbox(Path file) throws IOException { + this.file = file.toAbsolutePath().normalize(); + load(); + } + + synchronized boolean offer(String server, JsonEnvelope envelope) { + String key = key(server, envelope); + if (key == null) return false; + if (entries.containsKey(key)) return true; + if (entries.size() >= MAX_ENTRIES) return false; + String record = addRecord(server, envelope); + if (!prepareAppend(record) || !append(record)) return false; + entries.put(key, new Entry(server, envelope, false)); + journalRecords++; + return true; + } + + synchronized boolean acknowledgeCompletion(String server, UUID voteId, String subChannel) { + if (voteId == null || server == null || subChannel == null) return false; + String key = normalized(server) + '|' + subChannel + '|' + voteId; + Entry entry = entries.get(key); + if (entry == null) return false; + if (entry.awaitingReceiptRelease()) return true; + String record = COMPLETED + '\t' + encode(key) + '\n'; + if (!prepareAppend(record) || !append(record)) return false; + entries.put(key, new Entry(entry.server(), entry.envelope(), true)); + journalRecords++; + return true; + } + + synchronized boolean acknowledgeReceiptRelease(String server, UUID voteId, String subChannel) { + if (voteId == null || server == null || subChannel == null) return false; + String key = normalized(server) + '|' + subChannel + '|' + voteId; + Entry entry = entries.get(key); + if (entry == null || !entry.awaitingReceiptRelease()) return false; + return remove(key); + } + + private boolean remove(String key) { + if (entries.size() == 1) { + try { + if (!DurableFiles.deleteIfExists(file) && Files.exists(file)) return false; + } catch (IOException failure) { + return false; + } + entries.clear(); + journalRecords = 0; + return true; + } + String record = REMOVE + '\t' + encode(key) + '\n'; + if (!prepareAppend(record) || !append(record)) return false; + entries.remove(key); + journalRecords++; + return true; + } + + synchronized List snapshot() { + return new ArrayList<>(entries.values()); + } + + synchronized List pendingVotes() { + return entries.values().stream().filter(entry -> !entry.awaitingReceiptRelease()).toList(); + } + + synchronized List pendingReceiptReleases() { + return entries.values().stream().filter(Entry::awaitingReceiptRelease).toList(); + } + + synchronized int size() { + return entries.size(); + } + + private void load() throws IOException { + if (!Files.exists(file, LinkOption.NOFOLLOW_LINKS)) return; + if (Files.isSymbolicLink(file) || !Files.isRegularFile(file, LinkOption.NOFOLLOW_LINKS)) { + throw new IOException("Vote delivery outbox is not a regular file"); + } + if (Files.size(file) > MAX_FILE_BYTES) throw new IOException("Vote delivery outbox exceeds size limit"); + String content = Files.readString(file, StandardCharsets.UTF_8); + String[] lines = content.split("\\n", -1); + if (lines.length == 0 || !HEADER.equals(lines[0])) throw new IOException("Unsupported vote delivery outbox"); + boolean unterminatedTail = !content.endsWith("\n"); + int completeLineLimit = unterminatedTail ? lines.length - 1 : lines.length; + for (int index = 1; index < completeLineLimit; index++) { + if (lines[index].isBlank()) continue; + String[] parts = lines[index].split("\\t", 3); + try { + if (parts.length == 3 && ADD.equals(parts[0])) { + String server = decode(parts[1]); + JsonEnvelope envelope = JsonEnvelopeCodec.decode(decode(parts[2])); + String key = key(server, envelope); + if (key == null) throw new IllegalArgumentException("Invalid vote envelope"); + entries.put(key, new Entry(server, envelope, false)); + } else if (parts.length == 2 && COMPLETED.equals(parts[0])) { + String key = decode(parts[1]); + Entry entry = entries.get(key); + if (entry == null) throw new IllegalArgumentException("Completion without vote"); + entries.put(key, new Entry(entry.server(), entry.envelope(), true)); + } else if (parts.length == 2 && REMOVE.equals(parts[0])) { + entries.remove(decode(parts[1])); + } else throw new IllegalArgumentException("Unknown journal record"); + } catch (RuntimeException malformed) { + throw new IOException("Malformed vote delivery outbox entry", malformed); + } + journalRecords++; + if (entries.size() > MAX_ENTRIES) throw new IOException("Vote delivery outbox exceeds entry limit"); + } + if (unterminatedTail && !compact()) throw new IOException("Unable to repair vote delivery outbox"); + } + + private boolean prepareAppend(String record) { + try { + long projectedBytes = (Files.exists(file) ? Files.size(file) : HEADER.length() + 1L) + + record.getBytes(StandardCharsets.UTF_8).length; + boolean shouldCompact = journalRecords > entries.size() * 2 + 64; + if ((projectedBytes > MAX_FILE_BYTES || shouldCompact) && compact()) { + projectedBytes = Files.size(file) + record.getBytes(StandardCharsets.UTF_8).length; + } + return projectedBytes <= MAX_FILE_BYTES; + } catch (IOException failure) { + return false; + } + } + + private boolean append(String record) { + try { + Files.createDirectories(file.getParent()); + if (!Files.exists(file, LinkOption.NOFOLLOW_LINKS)) { + Path staged = file.resolveSibling(file.getFileName() + ".tmp"); + Files.writeString(staged, HEADER + '\n' + record, StandardCharsets.UTF_8, + StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE); + DurableFiles.publishStagedFile(staged, file); + } else { + Files.writeString(file, record, StandardCharsets.UTF_8, StandardOpenOption.APPEND, + StandardOpenOption.WRITE, LinkOption.NOFOLLOW_LINKS); + DurableFiles.forceFile(file); + } + return true; + } catch (IOException failure) { + return false; + } + } + + private boolean compact() { + try { + StringBuilder text = new StringBuilder(HEADER).append('\n'); + for (Entry entry : entries.values()) { + text.append(addRecord(entry.server(), entry.envelope())); + if (entry.awaitingReceiptRelease()) { + text.append(COMPLETED).append('\t').append(encode(key(entry.server(), entry.envelope()))).append('\n'); + } + } + Path staged = file.resolveSibling(file.getFileName() + ".tmp"); + Files.writeString(staged, text, StandardCharsets.UTF_8, StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE); + DurableFiles.publishStagedFile(staged, file); + journalRecords = entries.size() + (int) entries.values().stream() + .filter(Entry::awaitingReceiptRelease).count(); + return true; + } catch (IOException failure) { + return false; + } + } + + private static String addRecord(String server, JsonEnvelope envelope) { + return ADD + '\t' + encode(server) + '\t' + encode(JsonEnvelopeCodec.encode(envelope)) + '\n'; + } + + private static String key(String server, JsonEnvelope envelope) { + if (server == null || server.isBlank() || envelope == null) return null; + String subChannel = envelope.getSubChannel(); + if (!VotingPluginWire.SUB_VOTE.equals(subChannel) + && !VotingPluginWire.SUB_VOTE_ONLINE.equals(subChannel)) return null; + String voteId = envelope.getFields().get(VotingPluginWire.K_VOTE_ID); + try { + return normalized(server) + '|' + subChannel + '|' + UUID.fromString(voteId); + } catch (RuntimeException invalid) { + return null; + } + } + + private static String normalized(String value) { + return value.trim().toLowerCase(Locale.ROOT); + } + + private static String encode(String value) { + return Base64.getUrlEncoder().withoutPadding().encodeToString(value.getBytes(StandardCharsets.UTF_8)); + } + + private static String decode(String value) { + return new String(Base64.getUrlDecoder().decode(value), StandardCharsets.UTF_8); + } +} diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginProxy.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginProxy.java index fd917df8b..7aa116ed7 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginProxy.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginProxy.java @@ -129,6 +129,9 @@ private VoteRetryException() { private final Map liveVoteRetries = new LinkedHashMap<>(); private final Map multiProxyVoteRetries = new LinkedHashMap<>(); private final LinkedHashMap completedMultiProxyVotes = new LinkedHashMap<>(); + private final Set reliableVoteDeliveryServers = ConcurrentHashMap.newKeySet(); + private final Set legacyVoteDeliveryServers = ConcurrentHashMap.newKeySet(); + private ReliableVoteDeliveryOutbox reliableVoteDeliveryOutbox; // Set only after all replacement gates have succeeded. Vote entry points are // synchronized, so no new side-effecting vote can race the handoff window. private boolean runtimeReplacementPrepared; @@ -196,6 +199,7 @@ public void run() { } } private static final long PRESENCE_MAINTENANCE_INTERVAL_SECONDS = 30L; + private static final long VOTE_DELIVERY_CAPABILITY_PROBE_SECONDS = 60L; private static final long PRESENCE_BACKEND_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(90); private static final long CONTROL_ENROLLMENT_MIN_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(10); private static final long CONTROL_ENROLLMENT_CHALLENGE_TTL_NANOS = TimeUnit.MINUTES.toNanos(1); @@ -852,17 +856,148 @@ protected boolean sendVoteEnvelopeAccepted(String server, int delay, JsonEnvelop protected boolean sendVoteEnvelopeAccepted(String server, int delay, JsonEnvelope envelope, OfflineBungeeVote cachedVote) { + if (supportsReliableVoteDelivery(server)) { + ReliableVoteDeliveryOutbox outbox = reliableVoteDeliveryOutbox; + if (outbox == null || !outbox.offer(server, envelope)) { + logSevere("Unable to durably queue vote delivery for " + server); + return false; + } + JsonEnvelope requested = VotingPluginWire.requestVoteDeliveryAcknowledgement(envelope); + if (method == BungeeMethod.HTTP) sendGenericHttpEnvelope(server, requested); + else { + GlobalMessageProxyHandler handler = globalMessageProxyHandler; + if (handler != null) { + try { + handler.sendMessage(server, delay, requested); + } catch (RuntimeException failure) { + debug("Vote delivery remains queued after the immediate send failed for " + server); + } + } + } + return true; + } if (method == BungeeMethod.HTTP) { return sendHttpEnvelopeWithRecovery(server, envelope, cachedVote); } GlobalMessageProxyHandler handler = globalMessageProxyHandler; - if (handler == null) { - return false; - } + if (handler == null) return false; handler.sendMessage(server, delay, envelope); return true; } + private boolean supportsReliableVoteDelivery(String server) { + return server != null && reliableVoteDeliveryServers.contains(server.trim().toLowerCase(Locale.ROOT)); + } + + private void updateReliableVoteDeliveryCapability(String server, JsonEnvelope message) { + if (server == null || server.isBlank() || !isServerValid(server)) return; + String key = server.trim().toLowerCase(Locale.ROOT); + if (VotingPluginWire.advertisesVoteDeliveryAcknowledgement(message)) { + legacyVoteDeliveryServers.remove(key); + reliableVoteDeliveryServers.add(key); + retryReliableVoteDeliveries(server); + } else { + reliableVoteDeliveryServers.remove(key); + legacyVoteDeliveryServers.add(key); + retryReliableVoteDeliveries(server); + } + } + + private void retryReliableVoteDeliveries() { + retryReliableVoteDeliveries(null); + } + + private void retryReliableVoteDeliveries(String onlyServer) { + ReliableVoteDeliveryOutbox outbox = reliableVoteDeliveryOutbox; + GlobalMessageProxyHandler handler = globalMessageProxyHandler; + if (outbox == null || handler == null) return; + int delay = 1; + for (ReliableVoteDeliveryOutbox.Entry entry : outbox.snapshot()) { + if (onlyServer != null && !entry.server().equalsIgnoreCase(onlyServer)) continue; + try { + if (entry.awaitingReceiptRelease()) { + if (supportsReliableVoteDelivery(entry.server())) { + String voteId = entry.envelope().getFields().get(VotingPluginWire.K_VOTE_ID); + handler.sendMessage(entry.server(), delay++, VotingPluginWire.voteDeliveryReceiptRelease( + entry.server(), UUID.fromString(voteId), entry.envelope().getSubChannel())); + } + continue; + } + if (supportsReliableVoteDelivery(entry.server())) { + handler.sendMessage(entry.server(), delay++, + VotingPluginWire.requestVoteDeliveryAcknowledgement(entry.envelope())); + } else if (legacyVoteDeliveryServers.contains(entry.server().trim().toLowerCase(Locale.ROOT))) { + if (!sendProxyBroadcastEnvelopeNow(entry.server(), entry.envelope())) { + debug("Legacy vote delivery remains queued because the transport rejected it for " + + entry.server()); + continue; + } + delay++; + String voteId = entry.envelope().getFields().get(VotingPluginWire.K_VOTE_ID); + if (!outbox.acknowledgeCompletion(entry.server(), UUID.fromString(voteId), + entry.envelope().getSubChannel())) { + debug("Legacy vote delivery was accepted but remains queued until its release state is durable for " + + entry.server()); + } + } else { + continue; + } + } catch (RuntimeException failure) { + debug("Vote delivery retry remains queued for " + entry.server()); + } + } + } + + private void probeReliableVoteDeliveryCapabilities() { + if (method != BungeeMethod.PLUGINMESSAGING || globalMessageProxyHandler == null) return; + int delay = 1; + for (String server : getAllAvailableServers()) { + globalMessageProxyHandler.sendMessage(server, delay++, + VotingPluginWire.status(server, UUID.randomUUID())); + } + } + + private void handleVoteDeliveryAcknowledgement(JsonEnvelope message) { + if (!VotingPluginWire.advertisesVoteDeliveryAcknowledgement(message)) return; + String server = message.getFields().getOrDefault(VotingPluginWire.K_SERVER, ""); + String voteId = message.getFields().getOrDefault(VotingPluginWire.K_VOTE_ID, ""); + String subChannel = message.getFields().getOrDefault(VotingPluginWire.K_VOTE_DELIVERY_SUBCHANNEL, ""); + try { + if (!supportsReliableVoteDelivery(server)) return; + ReliableVoteDeliveryOutbox outbox = reliableVoteDeliveryOutbox; + UUID parsedVoteId = UUID.fromString(voteId); + if (outbox != null && !outbox.acknowledgeCompletion(server, parsedVoteId, subChannel)) { + debug("Ignored unmatched or unpersisted vote delivery acknowledgement from " + server); + } else if (outbox != null && globalMessageProxyHandler != null) { + try { + globalMessageProxyHandler.sendMessage(server, 1, + VotingPluginWire.voteDeliveryReceiptRelease(server, parsedVoteId, subChannel)); + } catch (RuntimeException failure) { + debug("Vote receipt release remains queued after the immediate send failed for " + server); + } + } + } catch (IllegalArgumentException invalidVoteId) { + debug("Ignored vote delivery acknowledgement with invalid vote ID from " + server); + } + } + + private void handleVoteDeliveryReceiptReleaseAcknowledgement(JsonEnvelope message) { + if (!VotingPluginWire.advertisesVoteDeliveryAcknowledgement(message)) return; + String server = message.getFields().getOrDefault(VotingPluginWire.K_SERVER, ""); + String voteId = message.getFields().getOrDefault(VotingPluginWire.K_VOTE_ID, ""); + String subChannel = message.getFields().getOrDefault(VotingPluginWire.K_VOTE_DELIVERY_SUBCHANNEL, ""); + try { + if (!supportsReliableVoteDelivery(server)) return; + ReliableVoteDeliveryOutbox outbox = reliableVoteDeliveryOutbox; + if (outbox != null && !outbox.acknowledgeReceiptRelease( + server, UUID.fromString(voteId), subChannel)) { + debug("Ignored unmatched or unpersisted vote receipt release acknowledgement from " + server); + } + } catch (IllegalArgumentException invalidVoteId) { + debug("Ignored vote receipt release acknowledgement with invalid vote ID from " + server); + } + } + public synchronized void checkCachedVotes(String server) { int delay = 1; if (isServerValid(server)) { @@ -1672,6 +1807,12 @@ public void debug1(Throwable e) { } }; voteCacheHandler.load(); + try { + reliableVoteDeliveryOutbox = new ReliableVoteDeliveryOutbox( + new File(getDataFolderPlugin(), "ProxyVoteDeliveryOutbox.dat").toPath()); + } catch (IOException failure) { + throw new IllegalStateException("Unable to load durable proxy vote delivery outbox", failure); + } method = retainHttpForPendingDeliveries(method); nonVotedPlayersCache = new NonVotedPlayersCache(getNonVotedCacheMySQLConfig(), @@ -1857,10 +1998,11 @@ public void onReceive(JsonEnvelope message) { globalMessageProxyHandler.addListener(new GlobalMessageListener(VotingPluginWire.SUB_BACKEND_STARTED) { @Override public void onReceive(JsonEnvelope message) { + String server = message.getFields().getOrDefault(VotingPluginWire.K_SERVER, ""); if (!method.supportsBackendPresence()) { + updateReliableVoteDeliveryCapability(server, message); return; } - String server = message.getFields().getOrDefault(VotingPluginWire.K_SERVER, ""); UUID backendIncarnationId = VotingPluginWire.readBackendIncarnationId(message); long backendStartedAt = VotingPluginWire.readBackendStartedAt(message); long presenceTimestamp = VotingPluginWire.readPresenceTimestamp(message); @@ -1869,6 +2011,7 @@ && isPresenceGenerationValid(backendIncarnationId, backendStartedAt, presenceTim VotingPluginWire.SUB_BACKEND_STARTED)) { if (backendPlayerPresenceTracker.backendStarted(server, backendIncarnationId, backendStartedAt, presenceTimestamp, System.currentTimeMillis())) { + updateReliableVoteDeliveryCapability(server, message); discardPendingPresenceHandoffs(server); pendingBackendRecoverySnapshots.add(presenceServerKey(server)); requestBackendPresenceSnapshot(server); @@ -1893,6 +2036,8 @@ && isPresenceGenerationValid(backendIncarnationId, backendStartedAt, presenceTim if (backendPlayerPresenceTracker.backendStopped(server, backendIncarnationId, backendStartedAt, presenceTimestamp, System.currentTimeMillis())) { discardPendingPresenceHandoffs(server); + reliableVoteDeliveryServers.remove(server.trim().toLowerCase(Locale.ROOT)); + legacyVoteDeliveryServers.remove(server.trim().toLowerCase(Locale.ROOT)); pendingBackendRecoverySnapshots.remove(presenceServerKey(server)); } } @@ -1912,8 +2057,10 @@ public void onReceive(JsonEnvelope message) { if (isPresenceServerValid(server, VotingPluginWire.SUB_BACKEND_HEARTBEAT) && isPresenceGenerationValid(backendIncarnationId, backendStartedAt, presenceTimestamp, VotingPluginWire.SUB_BACKEND_HEARTBEAT)) { - backendPlayerPresenceTracker.heartbeat(server, backendIncarnationId, backendStartedAt, - presenceTimestamp, System.currentTimeMillis()); + if (backendPlayerPresenceTracker.heartbeat(server, backendIncarnationId, backendStartedAt, + presenceTimestamp, System.currentTimeMillis())) { + updateReliableVoteDeliveryCapability(server, message); + } } } }); @@ -1955,10 +2102,26 @@ && isPresenceGenerationValid(snapshot.backendIncarnationId, snapshot.backendStar globalMessageProxyHandler.addListener(new GlobalMessageListener(VotingPluginWire.SUB_STATUS_OKAY) { @Override public void onReceive(JsonEnvelope message) { + updateReliableVoteDeliveryCapability(message.getFields().get(VotingPluginWire.K_SERVER), message); handleStatusOkay(message); } }); + globalMessageProxyHandler.addListener(new GlobalMessageListener(VotingPluginWire.SUB_VOTE_DELIVERY_ACK) { + @Override + public void onReceive(JsonEnvelope message) { + handleVoteDeliveryAcknowledgement(message); + } + }); + + globalMessageProxyHandler.addListener( + new GlobalMessageListener(VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE_ACK) { + @Override + public void onReceive(JsonEnvelope message) { + handleVoteDeliveryReceiptReleaseAcknowledgement(message); + } + }); + globalMessageProxyHandler.addListener(new GlobalMessageListener("voteupdate") { @Override public void onReceive(JsonEnvelope message) { @@ -1981,6 +2144,9 @@ public void onReceive(JsonEnvelope message) { loadTaskTimer(this::maintainBackendPresence, PRESENCE_MAINTENANCE_INTERVAL_SECONDS, PRESENCE_MAINTENANCE_INTERVAL_SECONDS); } + loadTaskTimer(this::retryReliableVoteDeliveries, 10L, 10L); + loadTaskTimer(this::probeReliableVoteDeliveryCapabilities, 1L, + VOTE_DELIVERY_CAPABILITY_PROBE_SECONDS); startControlServices(); // Open the listener last: backend callbacks can immediately reach routing, // presence, vote-log, multi-proxy, and Control-adjacent runtime helpers. diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginWire.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginWire.java index d4d781316..196daeaa9 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginWire.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/VotingPluginWire.java @@ -43,6 +43,9 @@ private VotingPluginWire() { public static final String SUB_VOTE_UPDATE = "VoteUpdate"; public static final String SUB_VOTE_DELAY_REJECTED = "VoteDelayRejected"; public static final String SUB_VOTE_BROADCAST = "VoteBroadcast"; + public static final String SUB_VOTE_DELIVERY_ACK = "VoteDeliveryAck"; + public static final String SUB_VOTE_DELIVERY_RECEIPT_RELEASE = "VoteDeliveryReceiptRelease"; + public static final String SUB_VOTE_DELIVERY_RECEIPT_RELEASE_ACK = "VoteDeliveryReceiptReleaseAck"; public static final String SUB_BUNGEE_TIME_CHANGE = "BungeeTimeChange"; public static final String SUB_STATUS = "Status"; @@ -110,6 +113,9 @@ private VotingPluginWire() { public static final String K_BUNGEE_BROADCAST = "bungeeBroadcast"; public static final String K_NUM = "num"; public static final String K_NUMBER_OF_VOTES = "numberOfVotes"; + public static final String K_VOTE_DELIVERY_ACK_VERSION = "voteDeliveryAckVersion"; + public static final String K_VOTE_DELIVERY_SUBCHANNEL = "voteDeliverySubchannel"; + public static final int VOTE_DELIVERY_ACK_VERSION = 2; /** Origin and receiving proxy names for reliable multi-proxy delivery. */ public static final String K_MULTI_PROXY_ORIGIN = "multiProxyOrigin"; public static final String K_MULTI_PROXY_RECIPIENT = "multiProxyRecipient"; @@ -204,6 +210,47 @@ public static JsonEnvelope votePartyBungee() { return base(SUB_VOTE_PARTY).build(); } + /** Requests an acknowledgement after the backend finishes an ordered vote. */ + public static JsonEnvelope requestVoteDeliveryAcknowledgement(JsonEnvelope envelope) { + JsonEnvelope.Builder builder = JsonEnvelope.builder(envelope.getSubChannel()).schema(envelope.getSchema()); + for (Map.Entry field : envelope.getFields().entrySet()) { + builder.put(field.getKey(), field.getValue()); + } + return builder.put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); + } + + public static boolean requestsVoteDeliveryAcknowledgement(JsonEnvelope envelope) { + return readInt(envelope.getFields(), K_VOTE_DELIVERY_ACK_VERSION, 0) >= VOTE_DELIVERY_ACK_VERSION; + } + + public static boolean advertisesVoteDeliveryAcknowledgement(JsonEnvelope envelope) { + return requestsVoteDeliveryAcknowledgement(envelope); + } + + public static JsonEnvelope voteDeliveryAcknowledgement(String server, UUID voteId, String voteSubchannel) { + return base(SUB_VOTE_DELIVERY_ACK).put(K_SERVER, safe(server)) + .put(K_VOTE_ID, voteId == null ? "" : voteId.toString()) + .put(K_VOTE_DELIVERY_SUBCHANNEL, safe(voteSubchannel)) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); + } + + /** Confirms durable proxy outbox completion so the backend can retire its receipt. */ + public static JsonEnvelope voteDeliveryReceiptRelease(String server, UUID voteId, String voteSubchannel) { + return base(SUB_VOTE_DELIVERY_RECEIPT_RELEASE).put(K_SERVER, safe(server)) + .put(K_VOTE_ID, voteId == null ? "" : voteId.toString()) + .put(K_VOTE_DELIVERY_SUBCHANNEL, safe(voteSubchannel)) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); + } + + /** Confirms durable backend receipt retirement to the originating proxy. */ + public static JsonEnvelope voteDeliveryReceiptReleaseAcknowledgement(String server, UUID voteId, + String voteSubchannel) { + return base(SUB_VOTE_DELIVERY_RECEIPT_RELEASE_ACK).put(K_SERVER, safe(server)) + .put(K_VOTE_ID, voteId == null ? "" : voteId.toString()) + .put(K_VOTE_DELIVERY_SUBCHANNEL, safe(voteSubchannel)) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); + } + public static JsonEnvelope status(String server) { return base(SUB_STATUS).put(K_SERVER, safe(server)).build(); } @@ -214,12 +261,14 @@ public static JsonEnvelope status(String server, UUID requestId) { } public static JsonEnvelope statusOkay(String server) { - return base(SUB_STATUS_OKAY).put(K_SERVER, safe(server)).build(); + return base(SUB_STATUS_OKAY).put(K_SERVER, safe(server)) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); } public static JsonEnvelope statusOkay(String server, UUID requestId) { return base(SUB_STATUS_OKAY).put(K_SERVER, safe(server)) - .put(K_REQUEST_ID, requestId == null ? "" : requestId.toString()).build(); + .put(K_REQUEST_ID, requestId == null ? "" : requestId.toString()) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); } public static JsonEnvelope serverName(String server) { @@ -313,7 +362,8 @@ public static JsonEnvelope backendStarted(String server, UUID backendIncarnation long presenceTimestamp) { return base(SUB_BACKEND_STARTED).put(K_SERVER, safe(server)).put(K_BACKEND_STARTED_AT, backendStartedAt) .put(K_BACKEND_INCARNATION_ID, backendIncarnationId == null ? "" : backendIncarnationId.toString()) - .put(K_PRESENCE_TIMESTAMP, presenceTimestamp).build(); + .put(K_PRESENCE_TIMESTAMP, presenceTimestamp) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); } public static JsonEnvelope backendStopped(String server) { @@ -343,7 +393,8 @@ public static JsonEnvelope backendHeartbeat(String server, UUID backendIncarnati long presenceTimestamp) { return base(SUB_BACKEND_HEARTBEAT).put(K_SERVER, safe(server)).put(K_BACKEND_STARTED_AT, backendStartedAt) .put(K_BACKEND_INCARNATION_ID, backendIncarnationId == null ? "" : backendIncarnationId.toString()) - .put(K_PRESENCE_TIMESTAMP, presenceTimestamp).build(); + .put(K_PRESENCE_TIMESTAMP, presenceTimestamp) + .put(K_VOTE_DELIVERY_ACK_VERSION, VOTE_DELIVERY_ACK_VERSION).build(); } public static JsonEnvelope presenceResyncRequest(String server, UUID requestId, long requestedAt) { diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandlerLifecycleTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandlerLifecycleTest.java index 2c62d70b0..b43f64c5d 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandlerLifecycleTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandlerLifecycleTest.java @@ -20,11 +20,13 @@ import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import java.lang.reflect.Field; import java.net.ServerSocket; import java.util.ArrayDeque; +import java.util.UUID; import java.nio.file.Path; import java.nio.file.Files; import java.util.concurrent.CompletableFuture; @@ -146,6 +148,123 @@ void voteAndVoteUpdateMessagesStayOrderedAcrossAsyncAndPlatformWork() throws Exc verify(scheduler, times(3)).runTaskAsynchronously(eq(plugin), any(Runnable.class)); } + @Test + void durableReceiptReleaseBypassesBlockedOrderedVote() throws Exception { + com.bencodez.votingplugin.VotingPluginMain plugin = mock(com.bencodez.votingplugin.VotingPluginMain.class); + BukkitScheduler scheduler = mock(BukkitScheduler.class); + when(plugin.getBukkitScheduler()).thenReturn(scheduler); + BackendProxyHandler handler = new BackendProxyHandler(plugin); + BackendProxyMessageRouter router = mock(BackendProxyMessageRouter.class); + setField(handler, "messageRouter", router); + handler.activateInboundMessages(); + + ArrayDeque asyncTasks = new ArrayDeque<>(); + doAnswer(invocation -> { + asyncTasks.addLast(invocation.getArgument(1)); + return null; + }).when(scheduler).runTaskAsynchronously(eq(plugin), any(Runnable.class)); + + JsonEnvelope blockedVote = JsonEnvelope.builder(VotingPluginWire.SUB_VOTE).build(); + JsonEnvelope release = VotingPluginWire.voteDeliveryReceiptRelease( + "survival", UUID.randomUUID(), VotingPluginWire.SUB_VOTE); + JsonEnvelope secondRelease = VotingPluginWire.voteDeliveryReceiptRelease( + "survival", UUID.randomUUID(), VotingPluginWire.SUB_VOTE); + when(router.hasDurableReceiptForRelease(release)).thenReturn(true); + when(router.hasDurableReceiptForRelease(secondRelease)).thenReturn(true); + doAnswer(invocation -> { + invocation.>getArgument(1) + .accept(OrderedVoteOutcome.RETRY); + return null; + }).when(router).handleOrderedVote(eq(release), any()); + + handler.dispatchIncomingAfterPublication(blockedVote, mock(Runnable.class)); + handler.dispatchIncomingAfterPublication(release, mock(Runnable.class)); + handler.dispatchIncomingAfterPublication(secondRelease, mock(Runnable.class)); + + assertEquals(2, asyncTasks.size(), "only one known release should use the bounded release worker"); + asyncTasks.removeLast().run(); + verify(router).handleOrderedVote(eq(release), any()); + @SuppressWarnings("unchecked") + ArrayDeque queued = (ArrayDeque) getField(handler, "orderedVoteDispatchQueue"); + assertEquals(java.util.List.of(blockedVote), java.util.List.copyOf(queued), + "capacity-blocked and concurrent releases must wait for the proxy retry outside the vote lane"); + } + + @Test + void queuedDurableReceiptReleaseCannotBlockLaterVote() throws Exception { + com.bencodez.votingplugin.VotingPluginMain plugin = mock(com.bencodez.votingplugin.VotingPluginMain.class); + BukkitScheduler scheduler = mock(BukkitScheduler.class); + when(plugin.getBukkitScheduler()).thenReturn(scheduler); + BackendProxyHandler handler = new BackendProxyHandler(plugin); + BackendProxyMessageRouter router = mock(BackendProxyMessageRouter.class); + setField(handler, "messageRouter", router); + + ArrayDeque asyncTasks = new ArrayDeque<>(); + doAnswer(invocation -> { + asyncTasks.addLast(invocation.getArgument(1)); + return null; + }).when(scheduler).runTaskAsynchronously(eq(plugin), any(Runnable.class)); + JsonEnvelope release = VotingPluginWire.voteDeliveryReceiptRelease( + "survival", UUID.randomUUID(), VotingPluginWire.SUB_VOTE); + JsonEnvelope laterVote = JsonEnvelope.builder(VotingPluginWire.SUB_VOTE).build(); + when(router.hasDurableReceiptForRelease(release)).thenReturn(true); + doAnswer(invocation -> { + invocation.>getArgument(1) + .accept(OrderedVoteOutcome.RETRY); + return null; + }).when(router).handleOrderedVote(eq(release), any()); + + @SuppressWarnings("unchecked") + ArrayDeque queued = (ArrayDeque) getField(handler, "orderedVoteDispatchQueue"); + queued.addLast(release); + queued.addLast(laterVote); + handler.activateInboundMessages(); + asyncTasks.removeFirst().run(); + + assertEquals(java.util.List.of(laterVote), java.util.List.copyOf(queued)); + assertEquals(2, asyncTasks.size(), + "release retry and the later vote should proceed independently after the stale local copy is removed"); + } + + @Test + void reliableVoteIsAcknowledgedOnlyAfterOrderedCompletion() throws Exception { + com.bencodez.votingplugin.VotingPluginMain plugin = mock(com.bencodez.votingplugin.VotingPluginMain.class); + BukkitScheduler scheduler = mock(BukkitScheduler.class); + BungeeSettings settings = mock(BungeeSettings.class); + when(plugin.getBukkitScheduler()).thenReturn(scheduler); + when(plugin.getBungeeSettings()).thenReturn(settings); + when(settings.getServer()).thenReturn("survival"); + BackendProxyHandler handler = new BackendProxyHandler(plugin); + BackendProxyMessageRouter router = mock(BackendProxyMessageRouter.class); + GlobalMessageHandler messages = mock(GlobalMessageHandler.class); + setField(handler, "messageRouter", router); + setField(handler, "globalMessageHandler", messages); + handler.activateInboundMessages(); + AtomicReference dispatch = new AtomicReference<>(); + AtomicReference> completion = new AtomicReference<>(); + doAnswer(invocation -> { + dispatch.set(invocation.getArgument(1)); + return null; + }).when(scheduler).runTaskAsynchronously(eq(plugin), any(Runnable.class)); + doAnswer(invocation -> { + completion.set(invocation.getArgument(1)); + return null; + }).when(router).handleOrderedVote(any(JsonEnvelope.class), any()); + UUID voteId = UUID.randomUUID(); + JsonEnvelope envelope = VotingPluginWire.requestVoteDeliveryAcknowledgement( + VotingPluginWire.vote("Player", UUID.randomUUID().toString(), "site", 10L, + true, true, "", voteId, false, false, 1, 1)); + + handler.dispatchIncomingAfterPublication(envelope, mock(Runnable.class)); + dispatch.get().run(); + verifyNoInteractions(messages); + completion.get().accept(OrderedVoteOutcome.COMPLETE); + + verify(messages).sendMessage(argThat(ack -> VotingPluginWire.SUB_VOTE_DELIVERY_ACK.equals(ack.getSubChannel()) + && voteId.toString().equals(ack.getFields().get(VotingPluginWire.K_VOTE_ID)) + && "survival".equals(ack.getFields().get(VotingPluginWire.K_SERVER)))); + } + @Test void orderedVoteBacklogSpillsPastBoundToDurableOverflow(@TempDir Path tempDir) throws Exception { com.bencodez.votingplugin.VotingPluginMain plugin = mock(com.bencodez.votingplugin.VotingPluginMain.class); diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCacheDurabilityTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCacheDurabilityTest.java new file mode 100644 index 000000000..0bc4d8b8c --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCacheDurabilityTest.java @@ -0,0 +1,145 @@ +package com.bencodez.votingplugin.backendproxy.cache; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.UUID; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class ProcessedVoteCacheDurabilityTest { + @TempDir + Path directory; + + @Test + void completedVoteRemainsDeduplicatedAfterBackendRestart() { + Path receipts = directory.resolve("receipts.dat"); + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache first = new ProcessedVoteCache(receipts); + + assertTrue(first.reserve(voteId)); + assertTrue(first.complete(voteId)); + assertTrue(first.hasDurableReceipt(voteId)); + assertFalse(new ProcessedVoteCache(receipts).reserve(voteId)); + } + + @Test + void reservationWithoutCompletionDoesNotSuppressRestartRetry() { + Path receipts = directory.resolve("receipts.dat"); + UUID voteId = UUID.randomUUID(); + assertTrue(new ProcessedVoteCache(receipts).reserve(voteId)); + + assertTrue(new ProcessedVoteCache(receipts).reserve(voteId)); + } + + @Test + void repairsTruncatedTailBeforeAppendingAnotherReceipt() throws Exception { + Path receipts = directory.resolve("receipts.dat"); + UUID firstId = UUID.randomUUID(); + ProcessedVoteCache first = new ProcessedVoteCache(receipts); + assertTrue(first.reserve(firstId)); + assertTrue(first.complete(firstId)); + Files.writeString(receipts, "truncated", StandardOpenOption.APPEND); + + ProcessedVoteCache repaired = new ProcessedVoteCache(receipts); + UUID secondId = UUID.randomUUID(); + assertTrue(repaired.reserve(secondId)); + assertTrue(repaired.complete(secondId)); + ProcessedVoteCache restarted = new ProcessedVoteCache(receipts); + assertFalse(restarted.reserve(firstId)); + assertFalse(restarted.reserve(secondId)); + } + + @Test + void discardsParseableUnterminatedReceiptTail() throws Exception { + Path receipts = directory.resolve("receipts.dat"); + UUID incompleteId = UUID.randomUUID(); + Files.writeString(receipts, "VP-VOTE-RECEIPTS-1\n" + incompleteId + "\t9"); + + ProcessedVoteCache repaired = new ProcessedVoteCache(receipts); + + assertTrue(repaired.reserve(incompleteId)); + assertTrue(Files.readString(receipts).endsWith("\n")); + } + + @Test + void receiptFailureFencesCompletedVotePastReservationExpiry() throws Exception { + Path unusableParent = directory.resolve("not-a-directory"); + Files.writeString(unusableParent, "file"); + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache cache = new ProcessedVoteCache(1L, unusableParent.resolve("receipts.dat")); + + assertTrue(cache.reserve(voteId)); + assertFalse(cache.complete(voteId)); + Thread.sleep(5L); + assertFalse(cache.reserve(voteId)); + } + + @Test + void proxyConfirmedReleaseLeavesRestartSafeTombstone() { + Path receipts = directory.resolve("receipts.dat"); + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache cache = new ProcessedVoteCache(receipts); + assertTrue(cache.reserve(voteId)); + assertTrue(cache.complete(voteId)); + + assertTrue(cache.releaseCompletedReceipt(voteId)); + + assertFalse(new ProcessedVoteCache(receipts).reserve(voteId)); + } + + @Test + void unknownReceiptReleaseLeavesRestartSafeTombstone() { + Path receipts = directory.resolve("receipts.dat"); + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache cache = new ProcessedVoteCache(receipts); + + assertFalse(cache.hasDurableReceipt(voteId)); + assertTrue(cache.releaseCompletedReceipt(voteId)); + assertTrue(cache.hasDurableReceipt(voteId)); + + assertFalse(new ProcessedVoteCache(receipts).reserve(voteId)); + } + + @Test + void completionHeadroomAllowsPrecedingVotesBeforeRelease() throws Exception { + DurableVoteReceiptStore store = new DurableVoteReceiptStore(directory.resolve("receipts.dat"), 1, 2, 1); + UUID first = UUID.randomUUID(); + UUID second = UUID.randomUUID(); + UUID third = UUID.randomUUID(); + UUID fourth = UUID.randomUUID(); + + assertTrue(store.complete(first) > 0L); + assertTrue(store.complete(second) > 0L); + assertTrue(store.complete(third) > 0L); + assertFalse(store.complete(fourth) > 0L); + assertTrue(store.release(first) > 0L); + assertTrue(store.complete(fourth) > 0L); + } + + @Test + void activeReceiptWaitsWhenTombstoneCapacityIsFull() throws Exception { + Path receipts = directory.resolve("receipts.dat"); + DurableVoteReceiptStore store = new DurableVoteReceiptStore(receipts, 1, 0, 1); + UUID active = UUID.randomUUID(); + assertTrue(store.complete(active) > 0L); + assertTrue(store.release(UUID.randomUUID()) > 0L); + + assertFalse(store.release(active) > 0L); + + new DurableVoteReceiptStore(receipts, 1, 0, 1); + } + + @Test + void expiredReleaseTombstoneIsReclaimedOnRestart() throws Exception { + Path receipts = directory.resolve("receipts.dat"); + UUID voteId = UUID.randomUUID(); + Files.writeString(receipts, "VP-VOTE-RECEIPTS-1\nR\t" + voteId + "\t1\n"); + + assertTrue(new ProcessedVoteCache(receipts).reserve(voteId)); + } +} diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouterTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouterTest.java index 4bce68c6d..266f65afa 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouterTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/backendproxy/messaging/BackendProxyMessageRouterTest.java @@ -2,6 +2,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyInt; @@ -11,6 +12,7 @@ import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -25,19 +27,24 @@ import com.bencodez.advancedcore.api.user.AdvancedCoreUser; import com.bencodez.advancedcore.api.user.usercache.UserDataManager; +import com.bencodez.advancedcore.AdvancedCoreConfigOptions; import com.bencodez.simpleapi.scheduler.BukkitScheduler; +import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope; +import com.bencodez.simpleapi.servercomm.global.GlobalMessageHandler; import com.bencodez.votingplugin.VotingPluginMain; import com.bencodez.votingplugin.backendproxy.cache.ProcessedVoteCache; import com.bencodez.votingplugin.backendproxy.messaging.BackendProxyMessageRouter.OrderedVoteOutcome; import com.bencodez.votingplugin.backendproxy.global.BackendGlobalDataSync; import com.bencodez.votingplugin.backendproxy.presence.BackendPresenceManager; import com.bencodez.votingplugin.backendproxy.voteparty.BackendVotePartySync; +import com.bencodez.votingplugin.proxy.BungeeMethod; import com.bencodez.votingplugin.proxy.VotingPluginWire; import com.bencodez.votingplugin.user.UserManager; import com.bencodez.votingplugin.user.VotingPluginUser; import com.bencodez.votingplugin.votesites.VoteSite; import com.bencodez.votingplugin.votesites.VoteSiteManager; import com.bencodez.votingplugin.config.BungeeSettings; +import com.bencodez.votingplugin.data.ServerData; class BackendProxyMessageRouterTest { @@ -176,7 +183,7 @@ void voteUpdateFailureAfterOfflineEffectRequestsQuarantine() { void voteRewardFailureRequestsQuarantineAfterVoteIdReservation() { UUID voteId = UUID.randomUUID(); ProcessedVoteCache cache = mock(ProcessedVoteCache.class); - when(cache.reserve(voteId)).thenReturn(true); + when(cache.reserve(voteId)).thenReturn(true, false); when(plugin.getBungeeSettings()).thenReturn(mock(BungeeSettings.class)); when(plugin.getVotingPluginUserManager().getVotingPluginUser(PLAYER_UUID, "Player")) .thenReturn(user); @@ -190,6 +197,96 @@ void voteRewardFailureRequestsQuarantineAfterVoteIdReservation() { VotingPluginWire.vote("Player", PLAYER_UUID.toString(), "known.example", LAST_VOTE_TIME, true, true, "", voteId, false, false, 1, 1), outcome::set)); assertEquals(OrderedVoteOutcome.QUARANTINE, outcome.get()); + + outcome.set(null); + voteRouter.handleOrderedVote( + VotingPluginWire.requestVoteDeliveryAcknowledgement( + VotingPluginWire.vote("Player", PLAYER_UUID.toString(), "known.example", LAST_VOTE_TIME, + true, true, "", voteId, false, false, 1, 1)), + outcome::set); + assertEquals(OrderedVoteOutcome.QUARANTINE, outcome.get()); + verify(cache, never()).complete(voteId); + } + + @Test + void reliableVoteRetriesReceiptPersistenceWithoutRepeatingEffectsInProcess() { + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache cache = mock(ProcessedVoteCache.class); + when(cache.reserve(voteId)).thenReturn(true, false); + when(cache.complete(voteId)).thenReturn(false, true); + when(cache.hasCompletedEffects(voteId)).thenReturn(true); + when(plugin.getBungeeSettings()).thenReturn(mock(BungeeSettings.class)); + when(plugin.getVotingPluginUserManager().getVotingPluginUser(PLAYER_UUID, "Player")) + .thenReturn(user); + when(plugin.getServerData()).thenReturn(mock(ServerData.class)); + BackendProxyMessageRouter voteRouter = new BackendProxyMessageRouter(plugin, + mock(BackendPresenceManager.class), mock(BackendGlobalDataSync.class), + mock(BackendVotePartySync.class), cache); + AtomicReference outcome = new AtomicReference<>(); + JsonEnvelope vote = VotingPluginWire.requestVoteDeliveryAcknowledgement( + VotingPluginWire.vote("Player", PLAYER_UUID.toString(), "known.example", LAST_VOTE_TIME, + true, true, "", voteId, false, false, 1, 1)); + + voteRouter.handleOrderedVote(vote, outcome::set); + assertEquals(OrderedVoteOutcome.RETRY, outcome.get()); + + voteRouter.handleOrderedVote(vote, outcome::set); + assertEquals(OrderedVoteOutcome.COMPLETE, outcome.get()); + verify(cache, times(2)).complete(voteId); + verify(user, times(1)).bungeeVotePluginMessaging(any(), anyLong(), any(), anyBoolean(), anyBoolean(), + anyBoolean(), anyInt()); + } + + @Test + void malformedReliableVoteIsQuarantinedWithoutCompletionReceipt() { + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache cache = mock(ProcessedVoteCache.class); + when(cache.reserve(voteId)).thenReturn(true); + when(plugin.getBungeeSettings()).thenReturn(mock(BungeeSettings.class)); + BackendProxyMessageRouter voteRouter = new BackendProxyMessageRouter(plugin, + mock(BackendPresenceManager.class), mock(BackendGlobalDataSync.class), + mock(BackendVotePartySync.class), cache); + AtomicReference outcome = new AtomicReference<>(); + JsonEnvelope vote = VotingPluginWire.requestVoteDeliveryAcknowledgement( + VotingPluginWire.vote("Player", "invalid-uuid", "known.example", LAST_VOTE_TIME, + true, true, "", voteId, false, false, 1, 1)); + + voteRouter.handleOrderedVote(vote, outcome::set); + + assertEquals(OrderedVoteOutcome.QUARANTINE, outcome.get()); + verify(cache, never()).complete(voteId); + verify(user, never()).bungeeVotePluginMessaging(any(), anyLong(), any(), anyBoolean(), anyBoolean(), + anyBoolean(), anyInt()); + } + + @Test + void receiptReleaseRunsInOrderedLaneAndAcknowledgesDurableRemoval() { + UUID voteId = UUID.randomUUID(); + ProcessedVoteCache cache = mock(ProcessedVoteCache.class); + when(cache.hasDurableReceipt(voteId)).thenReturn(true); + when(cache.releaseCompletedReceipt(voteId)).thenReturn(true); + AdvancedCoreConfigOptions options = mock(AdvancedCoreConfigOptions.class); + when(options.getServer()).thenReturn("survival"); + when(plugin.getOptions()).thenReturn(options); + GlobalMessageHandler messages = mock(GlobalMessageHandler.class); + BackendProxyMessageRouter voteRouter = new BackendProxyMessageRouter(plugin, + mock(BackendPresenceManager.class), mock(BackendGlobalDataSync.class), + mock(BackendVotePartySync.class), cache); + voteRouter.register(messages, BungeeMethod.REDIS); + AtomicReference outcome = new AtomicReference<>(); + + JsonEnvelope release = VotingPluginWire.voteDeliveryReceiptRelease( + "survival", voteId, VotingPluginWire.SUB_VOTE); + assertTrue(voteRouter.hasDurableReceiptForRelease(release)); + voteRouter.handleOrderedVote(release, outcome::set); + + assertEquals(OrderedVoteOutcome.COMPLETE, outcome.get()); + verify(cache).releaseCompletedReceipt(voteId); + org.mockito.ArgumentCaptor acknowledgement = org.mockito.ArgumentCaptor + .forClass(JsonEnvelope.class); + verify(messages).sendMessage(acknowledgement.capture()); + assertEquals(VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE_ACK, + acknowledgement.getValue().getSubChannel()); } } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutboxTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutboxTest.java new file mode 100644 index 000000000..3d5d05273 --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/ReliableVoteDeliveryOutboxTest.java @@ -0,0 +1,92 @@ +package com.bencodez.votingplugin.proxy; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.UUID; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope; + +class ReliableVoteDeliveryOutboxTest { + @TempDir + Path directory; + + @Test + void persistsUntilMatchingBackendAcknowledgesCompletion() throws Exception { + Path file = directory.resolve("outbox.dat"); + UUID voteId = UUID.randomUUID(); + JsonEnvelope vote = VotingPluginWire.vote("Player", UUID.randomUUID().toString(), "site", 10L, + true, true, "", voteId, false, false, 1, 1); + ReliableVoteDeliveryOutbox first = new ReliableVoteDeliveryOutbox(file); + + assertTrue(first.offer("Survival", vote)); + assertEquals(1, first.size()); + assertTrue(Files.isRegularFile(file)); + + ReliableVoteDeliveryOutbox restarted = new ReliableVoteDeliveryOutbox(file); + assertEquals(1, restarted.size()); + assertFalse(restarted.acknowledgeCompletion("creative", voteId, VotingPluginWire.SUB_VOTE)); + assertFalse(restarted.acknowledgeCompletion("survival", voteId, VotingPluginWire.SUB_VOTE_ONLINE)); + assertTrue(restarted.acknowledgeCompletion("survival", voteId, VotingPluginWire.SUB_VOTE)); + assertEquals(1, restarted.size()); + assertTrue(new ReliableVoteDeliveryOutbox(file).snapshot().get(0).awaitingReceiptRelease()); + assertTrue(restarted.acknowledgeReceiptRelease("survival", voteId, VotingPluginWire.SUB_VOTE)); + assertEquals(0, restarted.size()); + assertFalse(Files.exists(file)); + } + + @Test + void duplicateOfferKeepsOneStableDelivery() throws Exception { + UUID voteId = UUID.randomUUID(); + JsonEnvelope vote = VotingPluginWire.voteOnline("Player", UUID.randomUUID().toString(), "site", 10L, + true, true, "", voteId, false, false, 1, 1); + ReliableVoteDeliveryOutbox outbox = new ReliableVoteDeliveryOutbox(directory.resolve("outbox.dat")); + + assertTrue(outbox.offer("Survival", vote)); + assertTrue(outbox.offer("survival", vote)); + assertEquals(1, outbox.size()); + assertEquals(voteId.toString(), outbox.snapshot().get(0).envelope().getFields() + .get(VotingPluginWire.K_VOTE_ID)); + } + + @Test + void removalRecordSurvivesRestartWithoutDroppingOtherVotes() throws Exception { + Path file = directory.resolve("outbox.dat"); + UUID firstId = UUID.randomUUID(); + UUID secondId = UUID.randomUUID(); + ReliableVoteDeliveryOutbox outbox = new ReliableVoteDeliveryOutbox(file); + assertTrue(outbox.offer("survival", VotingPluginWire.vote("One", UUID.randomUUID().toString(), + "site", 10L, true, true, "", firstId, false, false, 1, 1))); + assertTrue(outbox.offer("survival", VotingPluginWire.vote("Two", UUID.randomUUID().toString(), + "site", 11L, true, true, "", secondId, false, false, 1, 1))); + assertTrue(outbox.acknowledgeCompletion("survival", firstId, VotingPluginWire.SUB_VOTE)); + assertTrue(outbox.acknowledgeReceiptRelease("survival", firstId, VotingPluginWire.SUB_VOTE)); + + ReliableVoteDeliveryOutbox restarted = new ReliableVoteDeliveryOutbox(file); + assertEquals(1, restarted.size()); + assertEquals(secondId.toString(), restarted.snapshot().get(0).envelope().getFields() + .get(VotingPluginWire.K_VOTE_ID)); + } + + @Test + void restartIgnoresOnlyATruncatedFinalAppend() throws Exception { + Path file = directory.resolve("outbox.dat"); + ReliableVoteDeliveryOutbox outbox = new ReliableVoteDeliveryOutbox(file); + assertTrue(outbox.offer("survival", VotingPluginWire.vote("One", UUID.randomUUID().toString(), + "site", 10L, true, true, "", UUID.randomUUID(), false, false, 1, 1))); + Files.writeString(file, "R\tc3Vydml2YWw", StandardOpenOption.APPEND); + + ReliableVoteDeliveryOutbox repaired = new ReliableVoteDeliveryOutbox(file); + assertEquals(1, repaired.size()); + assertTrue(repaired.offer("survival", VotingPluginWire.vote("Two", UUID.randomUUID().toString(), + "site", 11L, true, true, "", UUID.randomUUID(), false, false, 1, 1))); + assertEquals(2, new ReliableVoteDeliveryOutbox(file).size()); + } +} diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/VotingPluginProxyLifecycleTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/VotingPluginProxyLifecycleTest.java index 32b68ac8e..ac86e188d 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/VotingPluginProxyLifecycleTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/VotingPluginProxyLifecycleTest.java @@ -1,30 +1,169 @@ package com.bencodez.votingplugin.proxy; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; -import static org.junit.jupiter.api.Assertions.assertSame; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertThrows; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.nio.file.Path; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.UUID; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope; +import com.bencodez.simpleapi.servercomm.global.GlobalMessageProxyHandler; import com.bencodez.simpleapi.servercomm.sockets.ClientHandler; -import com.bencodez.votingplugin.tests.VotingPluginProxyTestImpl; -import com.bencodez.votingplugin.proxy.control.HostedControlManager; import com.bencodez.votingplugin.proxy.control.ControlConnector; +import com.bencodez.votingplugin.proxy.control.HostedControlManager; +import com.bencodez.votingplugin.tests.VotingPluginProxyTestImpl; class VotingPluginProxyLifecycleTest { + @Test + void completionAckTransitionsThroughDurableReceiptRelease(@TempDir Path directory) throws Exception { + VotingPluginProxyTestImpl proxy = new VotingPluginProxyTestImpl(); + GlobalMessageProxyHandler messages = mock(GlobalMessageProxyHandler.class); + UUID voteId = UUID.randomUUID(); + ReliableVoteDeliveryOutbox outbox = new ReliableVoteDeliveryOutbox(directory.resolve("outbox.dat")); + org.junit.jupiter.api.Assertions.assertTrue(outbox.offer("survival", VotingPluginWire.vote( + "Player", UUID.randomUUID().toString(), "site", 10L, true, true, "", voteId, + false, false, 1, 1))); + setField(proxy, "reliableVoteDeliveryOutbox", outbox); + setField(proxy, "globalMessageProxyHandler", messages); + @SuppressWarnings("unchecked") + Set reliable = (Set) field(proxy, "reliableVoteDeliveryServers"); + reliable.add("survival"); + Method completion = VotingPluginProxy.class.getDeclaredMethod( + "handleVoteDeliveryAcknowledgement", JsonEnvelope.class); + completion.setAccessible(true); + completion.invoke(proxy, VotingPluginWire.voteDeliveryAcknowledgement( + "survival", voteId, VotingPluginWire.SUB_VOTE)); + + assertEquals(1, outbox.size()); + org.junit.jupiter.api.Assertions.assertTrue(outbox.snapshot().get(0).awaitingReceiptRelease()); + ArgumentCaptor release = ArgumentCaptor.forClass(JsonEnvelope.class); + verify(messages).sendMessage(org.mockito.ArgumentMatchers.eq("survival"), + org.mockito.ArgumentMatchers.eq(1), release.capture()); + assertEquals(VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE, + release.getValue().getSubChannel()); + + Method released = VotingPluginProxy.class.getDeclaredMethod( + "handleVoteDeliveryReceiptReleaseAcknowledgement", JsonEnvelope.class); + released.setAccessible(true); + released.invoke(proxy, VotingPluginWire.voteDeliveryReceiptReleaseAcknowledgement( + "survival", voteId, VotingPluginWire.SUB_VOTE)); + assertEquals(0, outbox.size()); + } + + @Test + void preservesReceiptReleaseAfterLegacyDowngradeDelivery(@TempDir Path directory) throws Exception { + VotingPluginProxyTestImpl proxy = new VotingPluginProxyTestImpl(); + proxy.setMethod(BungeeMethod.PLUGINMESSAGING); + GlobalMessageProxyHandler messages = mock(GlobalMessageProxyHandler.class); + UUID voteId = UUID.randomUUID(); + JsonEnvelope vote = VotingPluginWire.vote("Player", UUID.randomUUID().toString(), "site", 10L, + true, true, "", voteId, false, false, 1, 1); + ReliableVoteDeliveryOutbox outbox = new ReliableVoteDeliveryOutbox(directory.resolve("outbox.dat")); + org.junit.jupiter.api.Assertions.assertTrue(outbox.offer("survival", vote)); + Field outboxField = VotingPluginProxy.class.getDeclaredField("reliableVoteDeliveryOutbox"); + outboxField.setAccessible(true); + outboxField.set(proxy, outbox); + Field messagesField = VotingPluginProxy.class.getDeclaredField("globalMessageProxyHandler"); + messagesField.setAccessible(true); + messagesField.set(proxy, messages); + Field legacyServers = VotingPluginProxy.class.getDeclaredField("legacyVoteDeliveryServers"); + legacyServers.setAccessible(true); + @SuppressWarnings("unchecked") + Set legacy = (Set) legacyServers.get(proxy); + Method retry = VotingPluginProxy.class.getDeclaredMethod("retryReliableVoteDeliveries", String.class); + retry.setAccessible(true); + retry.invoke(proxy, "survival"); + assertEquals(1, outbox.size()); + org.mockito.Mockito.verifyNoInteractions(messages); + + legacy.add("survival"); + proxy.setPluginMessageDeliveryResult(false); + retry.invoke(proxy, "survival"); + assertEquals(1, outbox.size()); + + proxy.setPluginMessageDeliveryResult(true); + retry.invoke(proxy, "survival"); + + assertEquals(1, outbox.size()); + org.junit.jupiter.api.Assertions.assertTrue(outbox.snapshot().get(0).awaitingReceiptRelease()); + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field field = VotingPluginProxy.class.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + + private static Object field(Object target, String name) throws Exception { + Field field = VotingPluginProxy.class.getDeclaredField(name); + field.setAccessible(true); + return field.get(target); + } + + @Test + void probesPluginMessagingBackendsForDeliveryCapability() throws Exception { + VotingPluginProxyTestImpl proxy = new VotingPluginProxyTestImpl(); + proxy.setMethod(BungeeMethod.PLUGINMESSAGING); + proxy.setAvailableServers("survival"); + GlobalMessageProxyHandler messages = mock(GlobalMessageProxyHandler.class); + Field messagesField = VotingPluginProxy.class.getDeclaredField("globalMessageProxyHandler"); + messagesField.setAccessible(true); + messagesField.set(proxy, messages); + Method probe = VotingPluginProxy.class.getDeclaredMethod("probeReliableVoteDeliveryCapabilities"); + probe.setAccessible(true); + + probe.invoke(proxy); + + org.mockito.ArgumentCaptor envelope = org.mockito.ArgumentCaptor.forClass(JsonEnvelope.class); + verify(messages).sendMessage(org.mockito.ArgumentMatchers.eq("survival"), + org.mockito.ArgumentMatchers.eq(1), envelope.capture()); + assertEquals(VotingPluginWire.SUB_STATUS, envelope.getValue().getSubChannel()); + org.junit.jupiter.api.Assertions.assertFalse(envelope.getValue().getFields() + .get(VotingPluginWire.K_REQUEST_ID).isBlank()); + } + + @Test + void capableHttpVoteRemainsInOutboxWhenImmediateTransportSendIsRejected(@TempDir Path directory) + throws Exception { + VotingPluginProxyTestImpl proxy = new VotingPluginProxyTestImpl(); + proxy.setMethod(BungeeMethod.HTTP); + proxy.setVoteEnvelopeDeliveryResult(false); + ReliableVoteDeliveryOutbox outbox = new ReliableVoteDeliveryOutbox(directory.resolve("outbox.dat")); + Field outboxField = VotingPluginProxy.class.getDeclaredField("reliableVoteDeliveryOutbox"); + outboxField.setAccessible(true); + outboxField.set(proxy, outbox); + Field reliableServers = VotingPluginProxy.class.getDeclaredField("reliableVoteDeliveryServers"); + reliableServers.setAccessible(true); + @SuppressWarnings("unchecked") + Set reliable = (Set) reliableServers.get(proxy); + reliable.add("survival"); + JsonEnvelope vote = VotingPluginWire.vote("Player", UUID.randomUUID().toString(), "site", 10L, + true, true, "", UUID.randomUUID(), false, false, 1, 1); + + org.junit.jupiter.api.Assertions.assertTrue(proxy.sendVoteEnvelopeAcceptedForTest("survival", 1, vote)); + + assertEquals(1, outbox.size()); + org.junit.jupiter.api.Assertions.assertTrue( + VotingPluginWire.requestsVoteDeliveryAcknowledgement(proxy.getLastVoteEnvelope())); + } @Test void retainsConnectorWhenOperationShutdownFails() throws Exception { diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginProxyTestImpl.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginProxyTestImpl.java index 610a77c1b..0499441bd 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginProxyTestImpl.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginProxyTestImpl.java @@ -437,12 +437,13 @@ public boolean sendProxyBroadcastImmediately(String server, JsonEnvelope envelop @Override protected boolean sendVoteEnvelopeAccepted(String server, int delay, JsonEnvelope envelope) { lastVoteEnvelope = envelope; - if (getMethod() == com.bencodez.votingplugin.proxy.BungeeMethod.HTTP) { - return voteEnvelopeDeliveryResult; - } return super.sendVoteEnvelopeAccepted(server, delay, envelope); } + public boolean sendVoteEnvelopeAcceptedForTest(String server, int delay, JsonEnvelope envelope) { + return sendVoteEnvelopeAccepted(server, delay, envelope); + } + public void setVoteEnvelopeDeliveryResult(boolean voteEnvelopeDeliveryResult) { this.voteEnvelopeDeliveryResult = voteEnvelopeDeliveryResult; } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginWireTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginWireTest.java index ff591c5e6..6b4b2f3cf 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginWireTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VotingPluginWireTest.java @@ -229,6 +229,34 @@ public void statusRoundTripCarriesCorrelationId() { assertEquals(requestId.toString(), request.getFields().get(VotingPluginWire.K_REQUEST_ID)); assertEquals(VotingPluginWire.SUB_STATUS_OKAY, response.getSubChannel()); assertEquals(requestId.toString(), response.getFields().get(VotingPluginWire.K_REQUEST_ID)); + assertTrue(VotingPluginWire.advertisesVoteDeliveryAcknowledgement(response)); + } + + @Test + public void voteDeliveryAcknowledgementIsAdditiveAndCorrelated() { + UUID voteId = UUID.randomUUID(); + JsonEnvelope vote = VotingPluginWire.vote("Player", UUID.randomUUID().toString(), "site", 10L, + true, true, "", voteId, false, false, 1, 1); + JsonEnvelope requested = VotingPluginWire.requestVoteDeliveryAcknowledgement(vote); + + assertFalse(VotingPluginWire.requestsVoteDeliveryAcknowledgement(vote)); + assertTrue(VotingPluginWire.requestsVoteDeliveryAcknowledgement(requested)); + assertEquals(voteId.toString(), requested.getFields().get(VotingPluginWire.K_VOTE_ID)); + + JsonEnvelope acknowledgement = VotingPluginWire.voteDeliveryAcknowledgement( + "survival", voteId, VotingPluginWire.SUB_VOTE); + assertEquals("survival", acknowledgement.getFields().get(VotingPluginWire.K_SERVER)); + assertEquals(voteId.toString(), acknowledgement.getFields().get(VotingPluginWire.K_VOTE_ID)); + assertEquals(VotingPluginWire.SUB_VOTE, + acknowledgement.getFields().get(VotingPluginWire.K_VOTE_DELIVERY_SUBCHANNEL)); + + JsonEnvelope release = VotingPluginWire.voteDeliveryReceiptRelease( + "survival", voteId, VotingPluginWire.SUB_VOTE); + JsonEnvelope releaseAck = VotingPluginWire.voteDeliveryReceiptReleaseAcknowledgement( + "survival", voteId, VotingPluginWire.SUB_VOTE); + assertEquals(VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE, release.getSubChannel()); + assertEquals(VotingPluginWire.SUB_VOTE_DELIVERY_RECEIPT_RELEASE_ACK, releaseAck.getSubChannel()); + assertTrue(VotingPluginWire.requestsVoteDeliveryAcknowledgement(release)); } @Test diff --git a/docs/proxy-vote-delivery.md b/docs/proxy-vote-delivery.md new file mode 100644 index 000000000..5928acfed --- /dev/null +++ b/docs/proxy-vote-delivery.md @@ -0,0 +1,53 @@ +# Proxy vote delivery + +Reward-bearing `Vote` and `VoteOnline` messages use an additive, versioned +acknowledgement protocol when both the proxy and backend support it. The backend +advertises `voteDeliveryAckVersion` through presence or status replies. Older +backends ignore the extra fields and retain the existing transport behavior. +Plugin messaging probes each available backend every minute because that +transport does not use backend presence heartbeats. + +Before a capable route reports acceptance to the existing vote pipeline, the +proxy writes the exact target server and envelope to +`ProxyVoteDeliveryOutbox.dat`. It retries that entry every ten seconds and after +capability discovery. The backend acknowledges only after its ordered vote lane +reports completion. A spilled message is acknowledged after removal from the +durable backend overflow queue. Before acknowledgement, the backend journals the +completed vote ID without time expiry so a delayed or lost acknowledgement +followed by backend restart does not repeat normal completed processing. The +bounded receipt journal fails closed at its capacity instead of evicting an ID +that may still have a proxy outbox entry. It reserves completion headroom larger +than the bounded ordered lane and separate space for release tombstones. A +release for an already-durable receipt runs through a bounded single-flight +worker outside the ordered vote lane because that receipt proves the vote +effects completed. Concurrent or capacity-blocked releases remain in the +proxy's durable outbox for its next retry; unknown releases remain ordered. The +proxy then durably transitions the +matching server, vote ID, and subchannel entry into a receipt-release phase. The +backend durably converts its completed receipt into a 24-hour post-release +tombstone and acknowledges that release; +only then does the proxy remove the outbox entry. Lost release messages and +acknowledgements remain retryable across either process restarting, so completed +receipts can be reclaimed without turning the journal bound into a lifetime cap. +If a backend generation stops advertising acknowledgements, the proxy drains +already-journaled entries once through the existing legacy send path and moves +each accepted entry into receipt release. This keeps rolling downgrades from +stranding accepted votes while retaining at-least-once behavior in case the previous capable backend +journaled completion before its acknowledgement was lost. Receipt-release +entries remain until a capable backend confirms retirement. + +This is an **at least once delivery guarantee**. Proxy shutdown, restart, a lost +send, or a lost acknowledgement leaves the outbox entry available for retry. +The backend vote ID cache and durable completion journal suppress ordinary and +restart-spanning duplicate retries. The bounded post-release tombstone also +fences transport retries that were already in flight when completion was +acknowledged. A backend process crash during reward side +effects, before the completion record is durable, can still lead to a repeated +attempt because reward execution and the receipt cannot be committed atomically. +Stronger exactly once reward execution would require a separate reward API and +storage design. + +HTTP retains its existing request recovery and also uses the completion outbox +after capability discovery. The generic outbox covers plugin messaging, Redis, +MQTT, MySQL, sockets, and HTTP. No reward data model or Bukkit vote ordering +changes.