From 07b96c4885f793ed921f71acd5760178aa091636 Mon Sep 17 00:00:00 2001 From: Ben Date: Tue, 13 Oct 2026 09:04:03 -0600 Subject: [PATCH 1/4] Fix malformed proxy cache identity recovery --- .../proxy/bungee/BungeeJsonVoteCache.java | 10 +- .../proxy/cache/VoteCacheDurability.java | 82 ++++ .../proxy/cache/VoteCacheHandler.java | 280 +++++++---- .../proxy/velocity/VelocityJsonVoteCache.java | 13 +- .../tests/CachedVoteBatchDurabilityTest.java | 442 +++++++++++++++++- 5 files changed, 729 insertions(+), 98 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/bungee/BungeeJsonVoteCache.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/bungee/BungeeJsonVoteCache.java index 06487f7bc..a5773de6c 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/bungee/BungeeJsonVoteCache.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/bungee/BungeeJsonVoteCache.java @@ -126,8 +126,12 @@ public synchronized void reload() { public synchronized void saveDurably() throws IOException { if (writesPaused) throw new IOException("Vote cache writes require a successful reload"); Path target = getStoragePath().toAbsolutePath().normalize(); + com.google.gson.JsonObject previous = null; if (journalLoaded) { - try { PendingPluginMessageVotePartyRewards.fromCacheJson(VoteCacheDurability.readJournal(target)); } + try { + previous = VoteCacheDurability.readJournal(target); + PendingPluginMessageVotePartyRewards.fromCacheJson(previous); + } catch (IOException | RuntimeException failure) { writesPaused = true; throw new IOException("Vote cache is unreadable; reload is required before saving", failure); @@ -140,6 +144,10 @@ public synchronized void saveDurably() throws IOException { try { Files.writeString(staged, new GsonBuilder().setPrettyPrinting().create().toJson(getConf()), StandardCharsets.UTF_8); + var candidate = VoteCacheDurability.readJournal(staged); + if (previous != null && hasEmergencyJournalMarker() + && VoteCacheDurability.preserveUnresolvedCacheRows(previous, candidate)) + Files.writeString(staged, candidate.toString(), StandardCharsets.UTF_8); PendingPluginMessageVotePartyRewards.fromCacheJson(VoteCacheDurability.readJournal(staged)); DurableFiles.publishStagedFile(staged, target); } finally { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java index 5b983440e..8dad0dd09 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java @@ -45,6 +45,88 @@ public static JsonObject readJournal(Path path) throws IOException { } } + /** Retains opaque recovery records while an unrelated healthy vote is updated or removed. */ + public static boolean preserveUnresolvedCacheRows(JsonObject previous, JsonObject candidate) { + boolean changed = false; + // JSON adapters may omit null/empty children. Successful updates of healthy votes + // must not erase another record's unresolved recovery evidence. + for (String lane : new String[] {"VoteCache", "OnlineCache"}) { + if (!previous.has(lane)) continue; + JsonElement owners = previous.get(lane); + if (!owners.isJsonObject()) { changed |= preserveValue(candidate, lane, owners); continue; } + for (var owner : owners.getAsJsonObject().entrySet()) { + if (!owner.getValue().isJsonObject()) { + changed |= preserveValue(objectChild(candidate, lane), owner.getKey(), owner.getValue()); + continue; + } + for (var row : owner.getValue().getAsJsonObject().entrySet()) + if (unresolvedRow(row.getValue())) + changed |= preserveValue(objectChild(objectChild(candidate, lane), owner.getKey()), + row.getKey(), row.getValue()); + } + } + return changed; + } + + private static JsonObject objectChild(JsonObject parent, String key) { + if (!parent.has(key) || !parent.get(key).isJsonObject()) parent.add(key, new JsonObject()); + return parent.getAsJsonObject(key); + } + + private static boolean unresolvedRow(JsonElement row) { + if (!row.isJsonObject()) return true; + JsonObject data = row.getAsJsonObject(); + if (hasUndecodableVoteFields(data)) return true; + if (!data.has("CachedBatchNumber") && !data.has("CachedBatchSize") + && !data.has("CachedBatchSourceRowId")) return false; + try { + for (String field : new String[] {"Name", "UUID", "Service", "Text"}) + if (!data.has(field) || !data.get(field).isJsonPrimitive() + || !data.getAsJsonPrimitive(field).isString()) return true; + if (!data.has("Real") || !data.get("Real").isJsonPrimitive() + || !data.getAsJsonPrimitive("Real").isBoolean() + || !data.has("Time") || !data.get("Time").isJsonPrimitive() + || !data.getAsJsonPrimitive("Time").isNumber()) return true; + new java.math.BigDecimal(data.get("Time").getAsString()).longValueExact(); + int[] values = new int[3]; + String[] fields = {"CachedBatchNumber", "CachedBatchSize", "CachedBatchSourceRowId"}; + for (int i = 0; i < fields.length; i++) { + if (!data.has(fields[i]) || !data.get(fields[i]).isJsonPrimitive() + || !data.getAsJsonPrimitive(fields[i]).isNumber()) return true; + values[i] = new java.math.BigDecimal(data.get(fields[i]).getAsString()).intValueExact(); + } + return values[2] < 0 || !(values[0] == 0 && values[1] == 0 + || values[0] > 0 && values[0] <= values[1]); + } catch (RuntimeException invalid) { return true; } + } + + private static boolean hasUndecodableVoteFields(JsonObject data) { + // Match fields which recovery actually reads, including optional delivery state. + // Do not classify unused/unknown null fields as broken legacy ownership. + for (String field : new String[] {"Name", "UUID", "Service", "Time", "Real", "Text", + "BroadcastForwarded", "ProxyBroadcastHandled", "BroadcastTargets", "BroadcastForwardedServers", + "HttpDeliveryIds", "HttpBroadcastDeliveryIds", "RewardDelivered", "WasOnlineKnown", + "DelayValidationKnown"}) + if (data.has(field) && !data.get(field).isJsonPrimitive()) return true; + String idField = data.has("VoteId") ? "VoteId" : "VoteID"; + if (data.has(idField)) { + if (!data.get(idField).isJsonPrimitive()) return true; + String id = data.get(idField).getAsString(); + try { if (!id.isEmpty()) java.util.UUID.fromString(id); } + catch (IllegalArgumentException invalid) { return true; } + } + if (data.has("WasOnlineKnown") && data.get("WasOnlineKnown").getAsBoolean() + && data.has("WasOnline") && !data.get("WasOnline").isJsonPrimitive()) return true; + return data.has("DelayValidationKnown") && data.get("DelayValidationKnown").getAsBoolean() + && data.has("DelayValidated") && !data.get("DelayValidated").isJsonPrimitive(); + } + + private static boolean preserveValue(JsonObject parent, String key, JsonElement value) { + if (value.equals(parent.get(key))) return false; + parent.add(key, value.deepCopy()); + return true; + } + /** Enforces JSON lexical rules which older Gson non-lenient readers leave permissive. */ private static final class StrictJsonInput extends java.io.FilterReader { private boolean firstCharacter = true; diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java index 6ed779872..1cce7eeb1 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java @@ -66,6 +66,8 @@ public synchronized int diagnosticPendingPersistenceVoteCount() { } private boolean jsonStorageQuarantined; + // Only a complete handler recovery may clear uncertain SQL/journal ownership. + private boolean unresolvedSqlBatchIdentity; /** Read-only health state; quarantined JSON mutations still require successful recovery. */ public synchronized boolean isJsonStorageQuarantined() { @@ -229,7 +231,7 @@ public synchronized boolean prepareCachedVoteBatch(String owner, boolean online, staged.put(vote, copy); } if (staged.isEmpty()) return true; - if (jsonStorage == null || jsonStorageQuarantined) return false; + if (jsonStorage == null || jsonStorageQuarantined || unresolvedSqlBatchIdentity) return false; synchronized (jsonStorage) { if (!ensureEmergencyJournalMarker()) return false; try { @@ -736,7 +738,6 @@ private OfflineBungeeVote findOnlineVoteSqlTwin(String uuid, OfflineBungeeVote e } private boolean matchesSqlBatchTwin(OfflineBungeeVote emergency, OfflineBungeeVote sql, int rowId) { - if (!emergency.isCachedVoteBatchValid()) return false; if (emergency.getVoteId() != null && sql.getVoteId() != null) return emergency.getVoteId().equals(sql.getVoteId()); return rowId > 0 && emergency.getCachedBatchSourceRowId() == rowId @@ -1428,6 +1429,7 @@ private String timeVoteCompletionIdentity(VoteTimeQueue vote) { * Loads vote cache from storage. */ public synchronized void load() { + if (useMySQL) unresolvedSqlBatchIdentity = true; if (jsonStorage != null) { try { jsonStorage.reload(); jsonStorageQuarantined = false; } catch (RuntimeException failure) { jsonStorageQuarantined = true; throw failure; } @@ -1443,84 +1445,94 @@ public synchronized void load() { } catch (SQLException failure) { throw new IllegalStateException("Unable to read proxy vote cache snapshot", failure); } - cachedVotes.clear(); - cachedOnlineVotes.clear(); - timeChangeQueue.clear(); - Set serverSourceRows = new LinkedHashSet<>(); - Set onlineSourceRows = new LinkedHashSet<>(); - for (ProxyVoteCacheTable.VoteRow row : serverRows) serverSourceRows.add(row.getId()); - for (ProxyOnlineVoteCacheTable.VoteRow row : onlineRows) onlineSourceRows.add(row.getId()); - // Load votes from MySQL - serverRows.forEach(voteRow -> { - OfflineBungeeVote vote = new OfflineBungeeVote(voteRow.getVoteId(), voteRow.getPlayerName(), - voteRow.getUuid(), voteRow.getService(), voteRow.getTime(), voteRow.isRealVote(), - voteRow.getText(), voteRow.isBroadcastForwarded(), voteRow.isProxyBroadcastHandled(), - VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastTargets()), - VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastForwardedServers()), - voteRow.isRewardDelivered(), OfflineBungeeVote.decodeHttpDeliveryIds(voteRow.getHttpDeliveryIds()), - OfflineBungeeVote.decodeHttpBroadcastDeliveryIds(voteRow.getHttpBroadcastDeliveryIds())); - if (voteRow.isWasOnlineKnown()) vote.setWasOnline(voteRow.isWasOnline()); - if (voteRow.getDelayValidation() >= 0) vote.setDelayValidated(voteRow.getDelayValidation() == 1); - else vote.clearDelayValidation(); - vote.setServerVoteCacheRowId(voteRow.getId()); - String server = voteRow.getServer(); - cachedVotes.putIfAbsent(server, new ArrayList<>()); - cachedVotes.get(server).add(vote); - }); - - // Load online votes from MySQL - onlineRows.forEach(voteRow -> { - OfflineBungeeVote vote = new OfflineBungeeVote(voteRow.getVoteId(), voteRow.getPlayerName(), - voteRow.getUuid(), voteRow.getService(), voteRow.getTime(), voteRow.isRealVote(), - voteRow.getText(), voteRow.isBroadcastForwarded(), voteRow.isProxyBroadcastHandled(), - VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastTargets()), - VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastForwardedServers()), - voteRow.isRewardDelivered(), OfflineBungeeVote.decodeHttpDeliveryIds(voteRow.getHttpDeliveryIds()), - OfflineBungeeVote.decodeHttpBroadcastDeliveryIds(voteRow.getHttpBroadcastDeliveryIds())); - if (voteRow.isWasOnlineKnown()) vote.setWasOnline(voteRow.isWasOnline()); - if (voteRow.getDelayValidation() >= 0) vote.setDelayValidated(voteRow.getDelayValidation() == 1); - else vote.clearDelayValidation(); - vote.setOnlineVoteCacheRowId(voteRow.getId()); - String player = vote.getUuid(); - cachedOnlineVotes.putIfAbsent(player, new ArrayList<>()); - cachedOnlineVotes.get(player).add(vote); - }); - - // Load timed votes from MySQL - ArrayList timedVotes = new ArrayList<>(); - timedRows.forEach(timedVoteRow -> { - VoteTimeQueue voteTimeQueue = new VoteTimeQueue(timedVoteRow.getVoteId(), timedVoteRow.getPlayerName(), - timedVoteRow.getService(), timedVoteRow.getTime(), timedVoteRow.isProxyBroadcastHandled(), - VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getBroadcastTargets()), - VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getBroadcastForwardedServers()), - timedVoteRow.getTotals(), timedVoteRow.isProcessed(), - timedVoteRow.isMultiProxyForwardingHandled(), timedVoteRow.getUuid(), - VoteTimeQueue.decodeHttpBroadcastDeliveryIds(timedVoteRow.getHttpBroadcastDeliveryIds())); - voteTimeQueue.setMultiProxyForwardingRequired(timedVoteRow.isMultiProxyForwardingRequired()); - voteTimeQueue.setMultiProxyCapabilityDiscoveryPending( - timedVoteRow.isMultiProxyCapabilityDiscoveryPending()); - voteTimeQueue.setVotePartyApplied(timedVoteRow.isVotePartyApplied()); - voteTimeQueue.setTotalsApplied(timedVoteRow.isTotalsApplied()); - voteTimeQueue.setRealVote(timedVoteRow.isRealVote()); - if (timedVoteRow.isWasOnlineKnown()) voteTimeQueue.setWasOnline(timedVoteRow.isWasOnline()); - if (timedVoteRow.getDelayValidation() >= 0) { - voteTimeQueue.setDelayValidated(timedVoteRow.getDelayValidation() == 1); - } else voteTimeQueue.clearDelayValidation(); - voteTimeQueue.setMultiProxyOrigin(timedVoteRow.getMultiProxyOrigin() == null ? "" - : timedVoteRow.getMultiProxyOrigin()); - voteTimeQueue.setMultiProxyCompletionPending(timedVoteRow.isMultiProxyCompletionPending()); - voteTimeQueue.setMultiProxyRecipients( - VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getMultiProxyRecipients())); - voteTimeQueue.setMultiProxyAcknowledgedServers( - VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getMultiProxyAcknowledgedServers())); - voteTimeQueue.setMultiProxyLegacyPendingRecipients( - VoteTimeQueue.decodeBroadcastForwardedServers( - timedVoteRow.getMultiProxyLegacyPendingRecipients())); - voteTimeQueue.setTimedVoteCacheRowId(timedVoteRow.getId()); - timedVotes.add(voteTimeQueue); - }); - timeChangeQueue.addAll(timedVotes); - loadJsonEmergencyVotes(serverSourceRows, onlineSourceRows); + ConcurrentHashMap> previousVotes = cachedVotes; + ConcurrentHashMap> previousOnlineVotes = cachedOnlineVotes; + Queue previousTimedVotes = timeChangeQueue; + cachedVotes = new ConcurrentHashMap<>(); + cachedOnlineVotes = new ConcurrentHashMap<>(); + timeChangeQueue = new ConcurrentLinkedQueue<>(); + try { + Set serverSourceRows = new LinkedHashSet<>(); + Set onlineSourceRows = new LinkedHashSet<>(); + for (ProxyVoteCacheTable.VoteRow row : serverRows) serverSourceRows.add(row.getId()); + for (ProxyOnlineVoteCacheTable.VoteRow row : onlineRows) onlineSourceRows.add(row.getId()); + // Load votes from MySQL + serverRows.forEach(voteRow -> { + OfflineBungeeVote vote = new OfflineBungeeVote(voteRow.getVoteId(), voteRow.getPlayerName(), + voteRow.getUuid(), voteRow.getService(), voteRow.getTime(), voteRow.isRealVote(), + voteRow.getText(), voteRow.isBroadcastForwarded(), voteRow.isProxyBroadcastHandled(), + VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastTargets()), + VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastForwardedServers()), + voteRow.isRewardDelivered(), OfflineBungeeVote.decodeHttpDeliveryIds(voteRow.getHttpDeliveryIds()), + OfflineBungeeVote.decodeHttpBroadcastDeliveryIds(voteRow.getHttpBroadcastDeliveryIds())); + if (voteRow.isWasOnlineKnown()) vote.setWasOnline(voteRow.isWasOnline()); + if (voteRow.getDelayValidation() >= 0) vote.setDelayValidated(voteRow.getDelayValidation() == 1); + else vote.clearDelayValidation(); + vote.setServerVoteCacheRowId(voteRow.getId()); + String server = voteRow.getServer(); + cachedVotes.putIfAbsent(server, new ArrayList<>()); + cachedVotes.get(server).add(vote); + }); + + // Load online votes from MySQL + onlineRows.forEach(voteRow -> { + OfflineBungeeVote vote = new OfflineBungeeVote(voteRow.getVoteId(), voteRow.getPlayerName(), + voteRow.getUuid(), voteRow.getService(), voteRow.getTime(), voteRow.isRealVote(), + voteRow.getText(), voteRow.isBroadcastForwarded(), voteRow.isProxyBroadcastHandled(), + VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastTargets()), + VoteTimeQueue.decodeBroadcastForwardedServers(voteRow.getBroadcastForwardedServers()), + voteRow.isRewardDelivered(), OfflineBungeeVote.decodeHttpDeliveryIds(voteRow.getHttpDeliveryIds()), + OfflineBungeeVote.decodeHttpBroadcastDeliveryIds(voteRow.getHttpBroadcastDeliveryIds())); + if (voteRow.isWasOnlineKnown()) vote.setWasOnline(voteRow.isWasOnline()); + if (voteRow.getDelayValidation() >= 0) vote.setDelayValidated(voteRow.getDelayValidation() == 1); + else vote.clearDelayValidation(); + vote.setOnlineVoteCacheRowId(voteRow.getId()); + String player = vote.getUuid(); + cachedOnlineVotes.putIfAbsent(player, new ArrayList<>()); + cachedOnlineVotes.get(player).add(vote); + }); + + // Load timed votes from MySQL + ArrayList timedVotes = new ArrayList<>(); + timedRows.forEach(timedVoteRow -> { + VoteTimeQueue voteTimeQueue = new VoteTimeQueue(timedVoteRow.getVoteId(), timedVoteRow.getPlayerName(), + timedVoteRow.getService(), timedVoteRow.getTime(), timedVoteRow.isProxyBroadcastHandled(), + VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getBroadcastTargets()), + VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getBroadcastForwardedServers()), + timedVoteRow.getTotals(), timedVoteRow.isProcessed(), + timedVoteRow.isMultiProxyForwardingHandled(), timedVoteRow.getUuid(), + VoteTimeQueue.decodeHttpBroadcastDeliveryIds(timedVoteRow.getHttpBroadcastDeliveryIds())); + voteTimeQueue.setMultiProxyForwardingRequired(timedVoteRow.isMultiProxyForwardingRequired()); + voteTimeQueue.setMultiProxyCapabilityDiscoveryPending( + timedVoteRow.isMultiProxyCapabilityDiscoveryPending()); + voteTimeQueue.setVotePartyApplied(timedVoteRow.isVotePartyApplied()); + voteTimeQueue.setTotalsApplied(timedVoteRow.isTotalsApplied()); + voteTimeQueue.setRealVote(timedVoteRow.isRealVote()); + if (timedVoteRow.isWasOnlineKnown()) voteTimeQueue.setWasOnline(timedVoteRow.isWasOnline()); + if (timedVoteRow.getDelayValidation() >= 0) { + voteTimeQueue.setDelayValidated(timedVoteRow.getDelayValidation() == 1); + } else voteTimeQueue.clearDelayValidation(); + voteTimeQueue.setMultiProxyOrigin(timedVoteRow.getMultiProxyOrigin() == null ? "" + : timedVoteRow.getMultiProxyOrigin()); + voteTimeQueue.setMultiProxyCompletionPending(timedVoteRow.isMultiProxyCompletionPending()); + voteTimeQueue.setMultiProxyRecipients( + VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getMultiProxyRecipients())); + voteTimeQueue.setMultiProxyAcknowledgedServers( + VoteTimeQueue.decodeBroadcastForwardedServers(timedVoteRow.getMultiProxyAcknowledgedServers())); + voteTimeQueue.setMultiProxyLegacyPendingRecipients( + VoteTimeQueue.decodeBroadcastForwardedServers( + timedVoteRow.getMultiProxyLegacyPendingRecipients())); + voteTimeQueue.setTimedVoteCacheRowId(timedVoteRow.getId()); + timedVotes.add(voteTimeQueue); + }); + timeChangeQueue.addAll(timedVotes); + unresolvedSqlBatchIdentity = !loadJsonEmergencyVotes(serverSourceRows, onlineSourceRows); + } catch (RuntimeException failure) { + cachedVotes = previousVotes; + cachedOnlineVotes = previousOnlineVotes; + timeChangeQueue = previousTimedVotes; + throw failure; + } } else { cachedVotes.clear(); @@ -1669,9 +1681,20 @@ public synchronized void load() { * unavailable. Stable vote IDs bind duplicates; legacy twins require the exact * persisted SQL source row and complete original vote identity. */ - private void loadJsonEmergencyVotes(Set serverSourceRows, Set onlineSourceRows) { - if (jsonStorage == null || !jsonStorage.hasEmergencyJournalMarker()) { - return; + private boolean loadJsonEmergencyVotes(Set serverSourceRows, Set onlineSourceRows) { + if (jsonStorage == null) return false; + if (!jsonStorage.hasEmergencyJournalMarker()) return true; + boolean resolved = true; + com.google.gson.JsonObject persistedJournal = null; + try { + if (jsonStorage.getStoragePath() != null) { + persistedJournal = VoteCacheDurability.readJournal(jsonStorage.getStoragePath()); + resolved = sqlJournalStructureIsValid(persistedJournal); + } + } + catch (IOException | RuntimeException invalid) { + resolved = false; + debug1(new IllegalStateException("Unable to classify SQL vote journal structure")); } List> orphanedServerTwins = new ArrayList<>(); List> orphanedOnlineTwins = new ArrayList<>(); @@ -1705,10 +1728,21 @@ private void loadJsonEmergencyVotes(Set serverSourceRows, Set continue; } for (String key : keys) { - OfflineBungeeVote vote = decodeJsonVote(jsonStorage.getServerVotes(server, key)); - if (vote == null) continue; + OfflineBungeeVote vote; + try { + DataNode data = persistedJournal == null ? jsonStorage.getServerVotes(server, key) + : persistedSqlJournalVote(persistedJournal, "VoteCache", server, key); + validateSqlJournalIdentity(data); + vote = decodeJsonVote(data); + } catch (RuntimeException invalid) { + resolved = false; + debug1(new IllegalStateException("Invalid SQL vote journal identity")); + continue; + } + if (vote == null) { resolved = false; continue; } vote.setServerVoteCacheJsonKey(key); OfflineBungeeVote sqlTwin = findServerVoteSqlTwin(server, vote); + if (!vote.isCachedVoteBatchValid()) resolved = false; if (sqlTwin != null) { sqlTwin.setServerVoteCacheJsonKey(key); sqlTwin.copyCachedVoteBatch(vote); @@ -1719,6 +1753,7 @@ else if (vote.getCachedBatchSourceRowId() > 0) { // A present or malformed unmatched source stays on disk, outside replay. if (vote.isCachedVoteBatchValid() && !serverSourceRows.contains(vote.getCachedBatchSourceRowId())) orphanedServerTwins.add(Map.entry(server, vote)); + else resolved = false; } else cachedVotes.computeIfAbsent(server, ignored -> new ArrayList<>()).add(vote); } } @@ -1731,10 +1766,21 @@ else if (vote.getCachedBatchSourceRowId() > 0) { continue; } for (String key : keys) { - OfflineBungeeVote vote = decodeJsonVote(jsonStorage.getOnlineVotes(player, key)); - if (vote == null) continue; + OfflineBungeeVote vote; + try { + DataNode data = persistedJournal == null ? jsonStorage.getOnlineVotes(player, key) + : persistedSqlJournalVote(persistedJournal, "OnlineCache", player, key); + validateSqlJournalIdentity(data); + vote = decodeJsonVote(data); + } catch (RuntimeException invalid) { + resolved = false; + debug1(new IllegalStateException("Invalid SQL vote journal identity")); + continue; + } + if (vote == null) { resolved = false; continue; } vote.setOnlineVoteCacheJsonKey(key); OfflineBungeeVote sqlTwin = findOnlineVoteSqlTwin(player, vote); + if (!vote.isCachedVoteBatchValid()) resolved = false; if (sqlTwin != null) { sqlTwin.setOnlineVoteCacheJsonKey(key); sqlTwin.copyCachedVoteBatch(vote); @@ -1743,22 +1789,65 @@ else if (vote.getCachedBatchSourceRowId() > 0) { else if (vote.getCachedBatchSourceRowId() > 0) { if (vote.isCachedVoteBatchValid() && !onlineSourceRows.contains(vote.getCachedBatchSourceRowId())) orphanedOnlineTwins.add(Map.entry(player, vote)); + else resolved = false; } else cachedOnlineVotes.computeIfAbsent(player, ignored -> new ArrayList<>()).add(vote); } } } } catch (RuntimeException e) { - debug1(e); + resolved = false; + debug1(new IllegalStateException("Unable to classify SQL vote journal")); } + // Retain the entire journal while any row cannot be safely reconstructed. + if (!resolved) return false; // Cleanup failure must not turn a confirmed removed SQL owner into a live vote. for (Entry orphan : orphanedServerTwins) { - try { removeJsonServerVoteDurably(orphan.getKey(), orphan.getValue()); } - catch (RuntimeException failure) { debug1(failure); } + try { if (!removeJsonServerVoteDurably(orphan.getKey(), orphan.getValue())) resolved = false; } + catch (RuntimeException failure) { resolved = false; debug1(failure); } } for (Entry orphan : orphanedOnlineTwins) { - try { removeJsonOnlineVoteDurably(orphan.getKey(), orphan.getValue()); } - catch (RuntimeException failure) { debug1(failure); } + try { if (!removeJsonOnlineVoteDurably(orphan.getKey(), orphan.getValue())) resolved = false; } + catch (RuntimeException failure) { resolved = false; debug1(failure); } + } + return resolved; + } + + private boolean sqlJournalStructureIsValid(com.google.gson.JsonObject document) { + // Configurate discards explicit JSON null children. Inspect persisted structure + // so those unresolved records cannot vanish from Velocity's recovery scan. + for (String lane : new String[] {"VoteCache", "OnlineCache"}) { + if (!document.has(lane)) continue; + var owners = document.get(lane); + if (!owners.isJsonObject()) return false; + for (var owner : owners.getAsJsonObject().entrySet()) { + if (!owner.getValue().isJsonObject()) return false; + for (var row : owner.getValue().getAsJsonObject().entrySet()) + if (!row.getValue().isJsonObject()) return false; + } } + return true; + } + + private DataNode persistedSqlJournalVote(com.google.gson.JsonObject document, String lane, + String owner, String key) { + // Classify raw values before an adapter can turn explicit null fields into absence. + var data = document.getAsJsonObject(lane).getAsJsonObject(owner).get(key); + return data == null ? null : new GsonDataNode(data); + } + + private void validateSqlJournalIdentity(DataNode data) { + if (data == null || !data.isObject()) throw new IllegalArgumentException("Invalid SQL vote journal row"); + // Old emergency rows without replay metadata retain their established optional fields. + if (!data.has("CachedBatchNumber") && !data.has("CachedBatchSize") + && !data.has("CachedBatchSourceRowId")) return; + for (String field : new String[] {"Name", "UUID", "Service", "Text"}) + if (!data.has(field) || !data.get(field).isPrimitive() + || !(data.get(field).asPrimitive() instanceof String)) + throw new IllegalArgumentException("Invalid SQL vote journal identity"); + if (!data.has("Real") || !(data.get("Real").asPrimitive() instanceof Boolean) + || !data.has("Time") || !(data.get("Time").asPrimitive() instanceof Number)) + throw new IllegalArgumentException("Invalid SQL vote journal identity"); + new java.math.BigDecimal(data.get("Time").asString()).longValueExact(); } private boolean ensureEmergencyJournalMarker() { @@ -1876,8 +1965,6 @@ private void restoreCachedVoteBatch(OfflineBungeeVote vote, DataNode data) { if (!data.has("CachedBatchNumber") && !data.has("CachedBatchSize") && !data.has("CachedBatchSourceRowId")) return; try { - int number = batchInteger(data, "CachedBatchNumber"); - int size = batchInteger(data, "CachedBatchSize"); int source = batchInteger(data, "CachedBatchSourceRowId"); if (source > 0) { for (String field : new String[] {"Name", "UUID", "Service", "Text"}) @@ -1889,8 +1976,11 @@ private void restoreCachedVoteBatch(OfflineBungeeVote vote, DataNode data) { throw new IllegalArgumentException("Invalid cached replay identity"); new java.math.BigDecimal(data.get("Time").asString()).longValueExact(); } - vote.setCachedVoteBatch(number, size); + if (source < 0) throw new IllegalArgumentException("Invalid cached replay source"); vote.setCachedBatchSourceRowId(source); + int number = batchInteger(data, "CachedBatchNumber"); + int size = batchInteger(data, "CachedBatchSize"); + vote.setCachedVoteBatch(number, size); if (!vote.isCachedVoteBatchValid()) throw new IllegalArgumentException("Invalid cached replay metadata"); vote.setCachedBatchDurable(vote.hasCachedVoteBatch()); } catch (RuntimeException invalid) { diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java index b48978fce..a59d9c451 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java @@ -24,6 +24,8 @@ import org.spongepowered.configurate.gson.GsonConfigurationLoader; +import com.google.gson.JsonObject; + /** * JSON-based vote cache implementation for Velocity proxy. */ @@ -95,8 +97,12 @@ public synchronized void reload() { public synchronized void saveDurably() throws IOException { if (writesPaused) throw new IOException("Vote cache writes require a successful reload"); Path target = getStoragePath().toAbsolutePath().normalize(); + JsonObject previous = null; if (journalLoaded) { - try { PendingPluginMessageVotePartyRewards.fromCacheJson(VoteCacheDurability.readJournal(target)); } + try { + previous = VoteCacheDurability.readJournal(target); + PendingPluginMessageVotePartyRewards.fromCacheJson(previous); + } catch (IOException | RuntimeException failure) { writesPaused = true; throw new IOException("Vote cache is unreadable; reload is required before saving", failure); @@ -115,6 +121,11 @@ public synchronized void saveDurably() throws IOException { } else { GsonConfigurationLoader.builder().path(staged).build().save(getConf()); } + JsonObject candidate = VoteCacheDurability.readJournal(staged); + if (previous != null && hasEmergencyJournalMarker() + && VoteCacheDurability.preserveUnresolvedCacheRows(previous, candidate)) { + Files.writeString(staged, candidate.toString(), StandardCharsets.UTF_8); + } PendingPluginMessageVotePartyRewards.fromCacheJson(VoteCacheDurability.readJournal(staged)); DurableFiles.publishStagedFile(staged, target); } finally { diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java index d18051ab3..fd198ec4a 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java @@ -15,6 +15,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import java.util.stream.Stream; import org.junit.jupiter.api.Test; @@ -33,6 +34,8 @@ import com.bencodez.votingplugin.proxy.cache.ProxyVoteCacheTable; import com.bencodez.votingplugin.proxy.cache.VoteCacheHandler; import com.bencodez.votingplugin.proxy.velocity.VelocityJsonVoteCache; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; /** Exercises real adapter publication/reload, with SQL represented only at its table boundary. */ @Execution(ExecutionMode.SAME_THREAD) @@ -309,7 +312,10 @@ Stream invalidBatchValuesAreRetainedButNeverPreparedForDelivery() { IVoteCache storage = cache(velocity); VoteCacheHandler writer = handler(storage); OfflineBungeeVote vote = vote(UUID.randomUUID()); add(writer, online, vote); Path file = storage.getStoragePath(); - Files.writeString(file, Files.readString(file).replace("\"CachedBatchNumber\": 0", "\"CachedBatchNumber\": 1.5")); + JsonObject document = JsonParser.parseString(Files.readString(file)).getAsJsonObject(); + journalRows(document, online).entrySet().forEach(entry -> + entry.getValue().getAsJsonObject().addProperty("CachedBatchNumber", 1.5)); + Files.writeString(file, document.toString()); VoteCacheHandler restarted = handler(cache(velocity)); restarted.load(); OfflineBungeeVote restored = votes(restarted, online).get(0); assertFalse(restored.isCachedVoteBatchValid()); @@ -586,6 +592,440 @@ Stream failedLegacyIdMigrationDoesNotPublishAnIdentityUntilItsJourn }); } + private record PreparedJournal(IVoteCache storage, List sqlVotes, + List identities, String cleanBytes) { } + + private PreparedJournal preparedJournal(boolean velocity, boolean online, boolean stableSqlId) throws Exception { + Files.deleteIfExists(directory.resolve("votecache.json")); + IVoteCache storage = cache(velocity); + List originals = List.of(vote(stableSqlId ? UUID.randomUUID() : null), vote(null)); + VoteCacheHandler writer = sqlHandler(storage, online, originals, List.of(41, 42)); + writer.load(); + assertTrue(writer.prepareCachedVoteBatch(owner(online), online, new ArrayList<>(votes(writer, online)), 2)); + return new PreparedJournal(storage, originals, + votes(writer, online).stream().map(OfflineBungeeVote::getVoteId).toList(), + Files.readString(storage.getStoragePath())); + } + + private JsonObject journalRows(JsonObject journal, boolean online) { + return journal.getAsJsonObject(online ? "OnlineCache" : "VoteCache").getAsJsonObject(owner(online)); + } + + private JsonObject sourceRow(JsonObject journal, boolean online, int source) { + return journalRows(journal, online).entrySet().stream().map(entry -> entry.getValue().getAsJsonObject()) + .filter(row -> row.get("CachedBatchSourceRowId").getAsInt() == source).findFirst().orElseThrow(); + } + + private OfflineBungeeVote sqlRow(VoteCacheHandler handler, boolean online, int row) { + return votes(handler, online).stream().filter(vote -> (online ? vote.getOnlineVoteCacheRowId() + : vote.getServerVoteCacheRowId()) == row).findFirst().orElseThrow(); + } + + private void verifyOnlyCheckedSqlReads(VoteCacheHandler handler) throws Exception { + verifyOnlyCheckedSqlReads(handler, 1); + } + + private void verifyOnlyCheckedSqlReads(VoteCacheHandler handler, int reads) throws Exception { + ProxyVoteCacheTable server = (ProxyVoteCacheTable) getField(handler, "voteCacheTable"); + ProxyOnlineVoteCacheTable online = (ProxyOnlineVoteCacheTable) getField(handler, "onlineVoteCacheTable"); + ProxyTimedVoteCacheTable timed = (ProxyTimedVoteCacheTable) getField(handler, "timedVoteCacheTable"); + verify(server, times(reads)).getAllVotesChecked(); + verify(online, times(reads)).getAllVotesChecked(); + verify(timed, times(reads)).getAllVotesChecked(); + verifyNoMoreInteractions(server, online, timed); + } + + @TestFactory + Stream malformedSqlBatchPositionsKeepIdentityAndSourceWithoutBecomingDurable() { + return lanes((velocity, online) -> { + List> corruptions = List.of( + row -> row.addProperty("CachedBatchNumber", "not-a-number"), + row -> row.addProperty("CachedBatchNumber", 1.5), + row -> row.addProperty("CachedBatchNumber", 2147483648L), + row -> row.remove("CachedBatchSize"), + row -> row.addProperty("CachedBatchSize", 0)); + for (Consumer corrupt : corruptions) { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + corrupt.accept(sourceRow(journal, online, 41)); + String malformed = journal.toString(); + Files.writeString(prepared.storage().getStoragePath(), malformed); + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + restarted.load(); + assertEquals(3, votes(restarted, online).size(), "malformed batch created a second reward owner"); + OfflineBungeeVote invalid = sqlRow(restarted, online, 41); + assertEquals(prepared.identities().get(0), invalid.getVoteId()); + assertEquals(41, invalid.getCachedBatchSourceRowId()); + assertFalse(invalid.isCachedVoteBatchValid()); + assertFalse(invalid.isCachedBatchDurable()); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(invalid), 1)); + OfflineBungeeVote clean = sqlRow(restarted, online, 42), fresh = sqlRow(restarted, online, 43); + assertEquals(prepared.identities().get(1), clean.getVoteId()); + assertEquals(2, clean.getCachedBatchNumber()); + assertTrue(clean.isCachedBatchDurable()); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(clean), 1)); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + assertNull(fresh.getVoteId()); + assertFalse(fresh.hasCachedVoteBatch()); + assertEquals(malformed, Files.readString(prepared.storage().getStoragePath())); + verifyOnlyCheckedSqlReads(restarted); + } + }); + } + + @TestFactory + Stream malformedRowsFenceNewBatchesAcrossLanesWhileCleanPreparedWorkContinues() { + return lanes((velocity, online) -> { + List> corruptions = List.of( + row -> row.remove("CachedBatchSourceRowId"), + row -> row.addProperty("CachedBatchSourceRowId", -41), + row -> row.addProperty("CachedBatchSourceRowId", "not-a-number"), + row -> row.addProperty("Real", "true"), + row -> row.addProperty("Time", "100"), + row -> row.addProperty("Time", 100.5), + row -> row.add("Name", new JsonObject())); + // The malformed row is in the opposite lane from the clean and fresh SQL work. + for (int variant = 0; variant < corruptions.size() + 2; variant++) { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject malformedRow = sourceRow(journal, online, 41).deepCopy(); + JsonObject oppositeRows = new JsonObject(), oppositeOwners = new JsonObject(); + if (variant < corruptions.size()) { + corruptions.get(variant).accept(malformedRow); + oppositeRows.add("9", malformedRow); + } else oppositeRows.add("9", JsonParser.parseString(variant == corruptions.size() ? "null" : "7")); + oppositeOwners.add(owner(!online), oppositeRows); + journal.add(online ? "VoteCache" : "OnlineCache", oppositeOwners); + String malformed = journal.toString(); + Files.writeString(prepared.storage().getStoragePath(), malformed); + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + restarted.load(); + OfflineBungeeVote clean = sqlRow(restarted, online, 42), fresh = sqlRow(restarted, online, 43); + assertTrue(clean.isCachedBatchDurable(), "unrelated malformed row blocked clean recovery, variant " + variant); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(clean), 1)); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1), + "malformed row did not fence the other lane, variant " + variant); + assertNull(fresh.getVoteId()); + assertFalse(fresh.hasCachedVoteBatch()); + assertEquals(malformed, Files.readString(prepared.storage().getStoragePath())); + verifyOnlyCheckedSqlReads(restarted); + } + }); + } + + @TestFactory + Stream malformedRecoveryFenceClearsOnlyAfterSuccessfulFullCheckedLoad() { + return lanes((velocity, online) -> { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + sourceRow(journal, online, 41).addProperty("CachedBatchNumber", "invalid"); + String malformed = journal.toString(); + Files.writeString(prepared.storage().getStoragePath(), malformed); + IVoteCache storage = cache(velocity); + VoteCacheHandler restarted = sqlHandler(storage, online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + restarted.load(); + OfflineBungeeVote fresh = sqlRow(restarted, online, 43); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + assertEquals(malformed, Files.readString(storage.getStoragePath())); + Files.writeString(storage.getStoragePath(), prepared.cleanBytes()); + storage.reload(); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1), + "adapter-only reload cleared the handler recovery fence"); + ProxyTimedVoteCacheTable timed = (ProxyTimedVoteCacheTable) getField(restarted, "timedVoteCacheTable"); + doThrow(new SQLException("last SQL lane unreadable")).when(timed).getAllVotesChecked(); + assertThrows(IllegalStateException.class, restarted::load); + assertSame(fresh, sqlRow(restarted, online, 43)); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1), + "partially checked SQL snapshot cleared the recovery fence"); + assertEquals(prepared.cleanBytes(), Files.readString(storage.getStoragePath())); + doReturn(List.of()).when(timed).getAllVotesChecked(); + restarted.load(); + OfflineBungeeVote restored = sqlRow(restarted, online, 41); + assertEquals(prepared.identities().get(0), restored.getVoteId()); + assertTrue(restored.isCachedBatchDurable()); + OfflineBungeeVote admitted = sqlRow(restarted, online, 43); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(admitted), 1)); + assertNotNull(admitted.getVoteId()); + assertEquals(43, admitted.getCachedBatchSourceRowId()); + assertTrue(admitted.isCachedBatchDurable()); + assertEquals(3, votes(restarted, online).size()); + verifyOnlyCheckedSqlReads(restarted, 3); + }); + } + + @TestFactory + Stream failedSqlReconstructionRetainsOwnedInstancesAndFenceUntilFullRecovery() { + return lanes((velocity, online) -> { + PreparedJournal prepared = preparedJournal(velocity, online, false); + IVoteCache storage = cache(velocity); + VoteCacheHandler restarted = sqlHandler(storage, online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + restarted.load(); + ArrayList owners = votes(restarted, online); + OfflineBungeeVote durable = sqlRow(restarted, online, 41), fresh = sqlRow(restarted, online, 43); + var timedOwners = restarted.getTimeChangeQueue(); + assertEquals(prepared.identities().get(0), durable.getVoteId()); + assertTrue(durable.isCachedBatchDurable()); + String before = Files.readString(storage.getStoragePath()); + verifyOnlyCheckedSqlReads(restarted); + + ProxyVoteCacheTable serverTable = (ProxyVoteCacheTable) getField(restarted, "voteCacheTable"); + ProxyOnlineVoteCacheTable onlineTable = (ProxyOnlineVoteCacheTable) getField(restarted, "onlineVoteCacheTable"); + ProxyTimedVoteCacheTable timedTable = (ProxyTimedVoteCacheTable) getField(restarted, "timedVoteCacheTable"); + // Every checked read succeeds, but a malformed row fails the subsequent reconstruction. + List healthyRows = online ? onlineTable.getAllVotesChecked() : serverTable.getAllVotesChecked(); + if (online) doReturn(java.util.Collections.singletonList(null)).when(onlineTable).getAllVotesChecked(); + else doReturn(java.util.Collections.singletonList(null)).when(serverTable).getAllVotesChecked(); + clearInvocations(serverTable, onlineTable, timedTable); + + assertThrows(RuntimeException.class, restarted::load); + assertSame(owners, votes(restarted, online)); + assertSame(durable, sqlRow(restarted, online, 41)); + assertSame(fresh, sqlRow(restarted, online, 43)); + assertSame(timedOwners, restarted.getTimeChangeQueue()); + assertEquals(prepared.identities(), owners.subList(0, 2).stream().map(OfflineBungeeVote::getVoteId).toList()); + assertEquals(41, durable.getCachedBatchSourceRowId()); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(durable), 1)); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + assertNull(fresh.getVoteId()); + assertFalse(fresh.hasCachedVoteBatch()); + assertEquals(before, Files.readString(storage.getStoragePath())); + + if (online) doReturn(healthyRows).when(onlineTable).getAllVotesChecked(); + else doReturn(healthyRows).when(serverTable).getAllVotesChecked(); + restarted.load(); + OfflineBungeeVote recovered = sqlRow(restarted, online, 41), admitted = sqlRow(restarted, online, 43); + assertEquals(durable.getVoteId(), recovered.getVoteId()); + assertEquals(41, recovered.getCachedBatchSourceRowId()); + assertTrue(recovered.isCachedBatchDurable()); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(admitted), 1)); + assertNotNull(admitted.getVoteId()); + assertTrue(admitted.isCachedBatchDurable()); + assertEquals(3, votes(restarted, online).size()); + verifyOnlyCheckedSqlReads(restarted, 2); + }); + } + + @TestFactory + Stream healthyUpdatesAndRemovalsPreserveMalformedRowsAndRestartFence() { + return lanes((velocity, online) -> { + for (int variant = 0; variant < 9; variant++) { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject malformedObject = sourceRow(journal, online, 41).deepCopy(); + if (variant == 2) malformedObject.add("Name", new JsonObject()); + if (variant == 3) malformedObject.addProperty("CachedBatchSize", 0); + if (variant == 4) malformedObject.add("VoteId", com.google.gson.JsonNull.INSTANCE); + if (variant == 5) malformedObject.add("RewardDelivered", com.google.gson.JsonNull.INSTANCE); + if (variant == 6) malformedObject.add("HttpDeliveryIds", com.google.gson.JsonNull.INSTANCE); + if (variant == 7) { + malformedObject.addProperty("WasOnlineKnown", true); + malformedObject.add("WasOnline", com.google.gson.JsonNull.INSTANCE); + } + if (variant == 8) malformedObject.addProperty("VoteId", "not-a-uuid"); + var malformed = variant < 2 ? JsonParser.parseString(variant == 0 ? "null" : "7") : malformedObject; + JsonObject oppositeRows = new JsonObject(), oppositeOwners = new JsonObject(); + oppositeRows.add("9", malformed); + oppositeOwners.add(owner(!online), oppositeRows); + journal.add(online ? "VoteCache" : "OnlineCache", oppositeOwners); + Files.writeString(prepared.storage().getStoragePath(), journal.toString()); + IVoteCache storage = spy(cache(velocity)); + OfflineBungeeVote freshSource = vote(null); + VoteCacheHandler recovering = sqlHandler(storage, online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), freshSource), List.of(41, 42, 43)); + recovering.load(); + OfflineBungeeVote clean = sqlRow(recovering, online, 42), fresh = sqlRow(recovering, online, 43); + assertTrue(clean.isCachedBatchDurable()); + assertFalse(recovering.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + ProxyVoteCacheTable serverTable = (ProxyVoteCacheTable) getField(recovering, "voteCacheTable"); + ProxyOnlineVoteCacheTable onlineTable = (ProxyOnlineVoteCacheTable) getField(recovering, "onlineVoteCacheTable"); + ProxyTimedVoteCacheTable timedTable = (ProxyTimedVoteCacheTable) getField(recovering, "timedVoteCacheTable"); + if (online) when(onlineTable.updateProxyBroadcastState(clean)).thenReturn(true); + else when(serverTable.updateProxyBroadcastState(clean, BACKEND)).thenReturn(true); + clean.setRewardDelivered(true); + assertTrue(online ? recovering.updateOnlineVote(PLAYER, clean) : recovering.updateServerVote(BACKEND, clean)); + JsonObject updated = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertTrue(journalRows(updated, !online).has("9"), "healthy update erased unresolved row, variant " + variant); + assertEquals(malformed, journalRows(updated, !online).get("9")); + assertTrue(sourceRow(updated, online, 42).get("RewardDelivered").getAsBoolean(), + "healthy delivery update was not actually persisted"); + verify(storage).saveDurably(); + + remove(recovering, online, clean); + JsonObject removed = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertTrue(journalRows(removed, !online).has("9"), "healthy removal erased unresolved row, variant " + variant); + assertEquals(malformed, journalRows(removed, !online).get("9")); + assertTrue(journalRows(removed, online).entrySet().stream().noneMatch(entry -> + entry.getValue().getAsJsonObject().get("VoteId").getAsString().equals(clean.getVoteId().toString()))); + verify(storage, times(2)).saveDurably(); + verify(serverTable).getAllVotesChecked(); + verify(onlineTable).getAllVotesChecked(); + verify(timedTable).getAllVotesChecked(); + if (online) { + verify(onlineTable).updateProxyBroadcastState(clean); + verify(onlineTable).tryRemoveVote(clean); + } else { + verify(serverTable).updateProxyBroadcastState(clean, BACKEND); + verify(serverTable).tryRemoveVote(clean, BACKEND); + } + verifyNoMoreInteractions(serverTable, onlineTable, timedTable); + + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, + List.of(prepared.sqlVotes().get(0), freshSource), List.of(41, 43)); + restarted.load(); + OfflineBungeeVote durable = sqlRow(restarted, online, 41), blocked = sqlRow(restarted, online, 43); + assertEquals(2, votes(restarted, online).size()); + assertEquals(prepared.identities().get(0), durable.getVoteId()); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(durable), 1)); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(blocked), 1), + "restart cleared the unresolved row fence after healthy writes, variant " + variant); + assertNull(blocked.getVoteId()); + assertFalse(blocked.hasCachedVoteBatch()); + JsonObject reopened = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertTrue(journalRows(reopened, !online).has("9")); + assertEquals(malformed, journalRows(reopened, !online).get("9")); + verifyOnlyCheckedSqlReads(restarted); + } + }); + } + + @TestFactory + Stream unusedNullFieldsAndIgnoredVoteIdAliasStillAllowLegacyBatchStaging() { + return lanes((velocity, online) -> { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject rows = journalRows(journal, online); + String otherKey = rows.entrySet().stream() + .filter(entry -> entry.getValue().getAsJsonObject().get("CachedBatchSourceRowId").getAsInt() == 42) + .map(java.util.Map.Entry::getKey).findFirst().orElseThrow(); + rows.remove(otherKey); + JsonObject legacy = sourceRow(journal, online, 41); + legacy.addProperty("CachedBatchNumber", 0); + legacy.addProperty("CachedBatchSize", 0); + legacy.addProperty("CachedBatchSourceRowId", 0); + legacy.add("UnknownOptionalField", com.google.gson.JsonNull.INSTANCE); + legacy.addProperty("WasOnlineKnown", false); + legacy.add("WasOnline", com.google.gson.JsonNull.INSTANCE); + legacy.addProperty("DelayValidationKnown", false); + legacy.add("DelayValidated", com.google.gson.JsonNull.INSTANCE); + // A valid canonical VoteId takes precedence over this unused old alias. + legacy.add("VoteID", com.google.gson.JsonNull.INSTANCE); + Files.writeString(prepared.storage().getStoragePath(), journal.toString()); + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, List.of(), List.of()); + restarted.load(); + assertEquals(1, votes(restarted, online).size()); + OfflineBungeeVote restored = votes(restarted, online).get(0); + assertEquals(prepared.identities().get(0), restored.getVoteId()); + assertTrue(restored.isCachedVoteBatchValid()); + assertFalse(restored.hasCachedVoteBatch()); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(restored), 1), + "unused null values blocked an otherwise valid legacy occurrence"); + assertEquals(prepared.identities().get(0), restored.getVoteId()); + assertTrue(restored.isCachedBatchDurable()); + assertEquals(1, restored.getCachedBatchNumber()); + assertEquals(1, restored.getCachedBatchSize()); + verifyOnlyCheckedSqlReads(restarted); + VoteCacheHandler next = sqlHandler(cache(velocity), online, List.of(), List.of()); + next.load(); + assertEquals(1, votes(next, online).size()); + assertEquals(restored.getVoteId(), votes(next, online).get(0).getVoteId()); + assertTrue(votes(next, online).get(0).isCachedBatchDurable()); + verifyOnlyCheckedSqlReads(next); + }); + } + + @TestFactory + Stream explicitNullBatchMetadataCannotBecomeLegacyWorkAfterFullReload() { + return lanes((velocity, online) -> { + List> nullFields = List.of(List.of("CachedBatchNumber"), List.of("CachedBatchSize"), + List.of("CachedBatchSourceRowId"), + List.of("CachedBatchNumber", "CachedBatchSize", "CachedBatchSourceRowId")); + for (List fields : nullFields) { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject malformed = sourceRow(journal, online, 41); + fields.forEach(field -> malformed.add(field, com.google.gson.JsonNull.INSTANCE)); + JsonObject retained = malformed.deepCopy(); + String malformedBytes = journal.toString(); + Files.writeString(prepared.storage().getStoragePath(), malformedBytes); + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + for (int reload = 0; reload < 2; reload++) { + restarted.load(); + var journalOwners = votes(restarted, online).stream() + .filter(vote -> prepared.identities().get(0).equals(vote.getVoteId())).toList(); + assertTrue(journalOwners.size() <= 1, "null metadata duplicated the original journal identity: " + fields); + for (OfflineBungeeVote journalOwner : journalOwners) { + assertFalse(journalOwner.isCachedVoteBatchValid(), "null metadata became valid legacy work: " + fields); + assertFalse(journalOwner.isCachedBatchDurable()); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(journalOwner), 1)); + } + OfflineBungeeVote originalSqlOwner = sqlRow(restarted, online, 41); + assertTrue(originalSqlOwner.getVoteId() == null + || prepared.identities().get(0).equals(originalSqlOwner.getVoteId())); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(originalSqlOwner), 1), + "null metadata allowed the legacy SQL owner to acquire a new identity: " + fields); + OfflineBungeeVote clean = sqlRow(restarted, online, 42), fresh = sqlRow(restarted, online, 43); + assertEquals(prepared.identities().get(1), clean.getVoteId()); + assertTrue(clean.isCachedBatchDurable()); + assertTrue(restarted.prepareCachedVoteBatch(owner(online), online, List.of(clean), 1)); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + assertNull(fresh.getVoteId()); + assertFalse(fresh.hasCachedVoteBatch()); + JsonObject raw = JsonParser.parseString(Files.readString(prepared.storage().getStoragePath())).getAsJsonObject(); + JsonObject rawOwner = journalRows(raw, online).entrySet().stream() + .map(entry -> entry.getValue().getAsJsonObject()) + .filter(row -> row.get("VoteId").getAsString().equals(prepared.identities().get(0).toString())) + .findFirst().orElseThrow(); + assertEquals(retained, rawOwner, "full reload normalized the unresolved null metadata: " + fields); + assertEquals(malformedBytes, Files.readString(prepared.storage().getStoragePath())); + } + verifyOnlyCheckedSqlReads(restarted, 2); + } + }); + } + + @TestFactory + Stream malformedPositionsBindExactStableSqlIdsButNeverReusedRowPayloads() { + return lanes((velocity, online) -> { + PreparedJournal prepared = preparedJournal(velocity, online, true); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + sourceRow(journal, online, 41).addProperty("CachedBatchNumber", 1.5); + String malformed = journal.toString(); + Files.writeString(prepared.storage().getStoragePath(), malformed); + // A stable identity still binds when the SQL physical row moved. + VoteCacheHandler stable = sqlHandler(cache(velocity), online, prepared.sqlVotes(), List.of(51, 42)); + stable.load(); + assertEquals(2, votes(stable, online).size()); + OfflineBungeeVote restored = sqlRow(stable, online, 51); + assertEquals(prepared.identities().get(0), restored.getVoteId()); + assertFalse(restored.isCachedVoteBatchValid()); + assertFalse(restored.isCachedBatchDurable()); + assertFalse(stable.prepareCachedVoteBatch(owner(online), online, List.of(restored), 1)); + assertEquals(malformed, Files.readString(prepared.storage().getStoragePath())); + verifyOnlyCheckedSqlReads(stable); + + // Reusing the old physical row for another occurrence must not claim the old identity. + OfflineBungeeVote replacement = new OfflineBungeeVote((UUID) null, "test-player", PLAYER, + "test-service", 100L, false, "other-totals"); + VoteCacheHandler reused = sqlHandler(cache(velocity), online, + List.of(replacement, prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + reused.load(); + OfflineBungeeVote replacementOwner = sqlRow(reused, online, 41); + assertNull(replacementOwner.getVoteId()); + assertFalse(replacementOwner.hasCachedVoteBatch()); + assertEquals(3, votes(reused, online).size(), "unmatched linked journal became a new reward owner"); + assertFalse(reused.prepareCachedVoteBatch(owner(online), online, List.of(replacementOwner), 1)); + assertFalse(reused.prepareCachedVoteBatch(owner(online), online, List.of(sqlRow(reused, online, 43)), 1)); + assertTrue(reused.prepareCachedVoteBatch(owner(online), online, List.of(sqlRow(reused, online, 42)), 1)); + assertEquals(malformed, Files.readString(prepared.storage().getStoragePath())); + verifyOnlyCheckedSqlReads(reused); + }); + } + private VoteCacheHandler sqlHandler(IVoteCache storage, boolean online, List votes, List rows) throws Exception { VoteCacheHandler handler = handler(storage); From 306854486b86653d6b805c54451562a1eaa7a4bf Mon Sep 17 00:00:00 2001 From: Ben Date: Tue, 13 Oct 2026 09:36:30 -0600 Subject: [PATCH 2/4] Preserve recovery journal keys and owner delivery progress --- .../proxy/cache/VoteCacheDurability.java | 53 +++++++++- .../proxy/cache/VoteCacheHandler.java | 26 ++++- .../proxy/velocity/VelocityJsonVoteCache.java | 3 +- .../tests/CachedVoteBatchDurabilityTest.java | 98 +++++++++++++++++++ .../tests/VoteCacheHandlerVoteIdTest.java | 8 +- 5 files changed, 182 insertions(+), 6 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java index 8dad0dd09..1c7506020 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java @@ -122,11 +122,60 @@ private static boolean hasUndecodableVoteFields(JsonObject data) { } private static boolean preserveValue(JsonObject parent, String key, JsonElement value) { - if (value.equals(parent.get(key))) return false; - parent.add(key, value.deepCopy()); + JsonElement current = parent.get(key); + JsonElement retained = value.deepCopy(); + if (sameRecoveryOwner(value, current)) { + // Keep corrupt identity/position evidence, but persist actual completion + // progress of the same owner instead of silently reverting that update. + JsonObject original = retained.getAsJsonObject(), updated = current.getAsJsonObject(); + for (String field : new String[] {"BroadcastForwarded", "ProxyBroadcastHandled", "BroadcastTargets", + "BroadcastForwardedServers", "HttpDeliveryIds", "HttpBroadcastDeliveryIds", "RewardDelivered", + "WasOnlineKnown", "WasOnline", "DelayValidationKnown", "DelayValidated"}) + if (updated.has(field)) original.add(field, updated.get(field).deepCopy()); + } + if (retained.equals(current)) return false; + parent.add(key, retained); return true; } + private static boolean sameRecoveryOwner(JsonElement previous, JsonElement candidate) { + if (candidate == null || !previous.isJsonObject() || !candidate.isJsonObject()) return false; + JsonObject left = previous.getAsJsonObject(), right = candidate.getAsJsonObject(); + try { + java.util.UUID leftId = recoveryVoteId(left), rightId = recoveryVoteId(right); + if (leftId != null || rightId != null) return leftId != null && leftId.equals(rightId); + if (!left.has("CachedBatchSourceRowId") || !right.has("CachedBatchSourceRowId") + || !left.get("CachedBatchSourceRowId").isJsonPrimitive() + || !right.get("CachedBatchSourceRowId").isJsonPrimitive() + || !left.getAsJsonPrimitive("CachedBatchSourceRowId").isNumber() + || !right.getAsJsonPrimitive("CachedBatchSourceRowId").isNumber()) return false; + int source = new java.math.BigDecimal(left.get("CachedBatchSourceRowId").getAsString()).intValueExact(); + if (source <= 0 || source != new java.math.BigDecimal(right.get("CachedBatchSourceRowId") + .getAsString()).intValueExact()) return false; + for (String field : new String[] {"Name", "UUID", "Service", "Text"}) + if (!left.has(field) || !right.has(field) || !left.get(field).isJsonPrimitive() + || !right.get(field).isJsonPrimitive() || !left.getAsJsonPrimitive(field).isString() + || !right.getAsJsonPrimitive(field).isString() || !left.get(field).equals(right.get(field))) return false; + return left.has("Real") && right.has("Real") && left.get("Real").isJsonPrimitive() + && right.get("Real").isJsonPrimitive() && left.getAsJsonPrimitive("Real").isBoolean() + && right.getAsJsonPrimitive("Real").isBoolean() && left.get("Real").equals(right.get("Real")) + && left.has("Time") && right.has("Time") && left.get("Time").isJsonPrimitive() + && right.get("Time").isJsonPrimitive() && left.getAsJsonPrimitive("Time").isNumber() + && right.getAsJsonPrimitive("Time").isNumber() + && new java.math.BigDecimal(left.get("Time").getAsString()).longValueExact() + == new java.math.BigDecimal(right.get("Time").getAsString()).longValueExact(); + } catch (RuntimeException invalid) { return false; } + } + + private static java.util.UUID recoveryVoteId(JsonObject data) { + String field = data.has("VoteId") ? "VoteId" : "VoteID"; + if (!data.has(field)) return null; + if (!data.get(field).isJsonPrimitive() || !data.getAsJsonPrimitive(field).isString()) + throw new IllegalArgumentException("Invalid persisted vote identity"); + String value = data.get(field).getAsString(); + return value.isEmpty() ? null : java.util.UUID.fromString(value); + } + /** Enforces JSON lexical rules which older Gson non-lenient readers leave permissive. */ private static final class StrictJsonInput extends java.io.FilterReader { private boolean firstCharacter = true; diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java index 1cce7eeb1..b30e4d64f 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java @@ -843,7 +843,7 @@ private boolean persistServerVoteToJsonLocked(String server, OfflineBungeeVote v return false; } try { - Collection keys = jsonStorage.getServerVotes(server); + Collection keys = reservedJournalKeys(server, false); int index = nextCacheIndex(keys); vote.setServerVoteCacheJsonKey(String.valueOf(index)); jsonStorage.addVote(server, index, vote); @@ -868,7 +868,7 @@ private boolean persistOnlineVoteToJsonLocked(String uuid, OfflineBungeeVote vot return false; } try { - Collection keys = jsonStorage.getOnlineVotes(uuid); + Collection keys = reservedJournalKeys(uuid, true); int index = nextCacheIndex(keys); vote.setOnlineVoteCacheJsonKey(String.valueOf(index)); jsonStorage.addVoteOnline(uuid, index, vote); @@ -879,6 +879,28 @@ private boolean persistOnlineVoteToJsonLocked(String uuid, OfflineBungeeVote vot } } + private Collection reservedJournalKeys(String owner, boolean online) { + Collection loaded = online ? jsonStorage.getOnlineVotes(owner) : jsonStorage.getServerVotes(owner); + Set keys = new LinkedHashSet<>(); + if (loaded != null) keys.addAll(loaded); + Path path = jsonStorage.getStoragePath(); + if (path == null) return keys; + try { + var document = VoteCacheDurability.readJournal(path); + String lane = online ? "OnlineCache" : "VoteCache"; + if (!document.has(lane)) return keys; + var owners = document.get(lane); + if (!owners.isJsonObject()) throw new IllegalStateException("Unresolved cache lane structure"); + if (!owners.getAsJsonObject().has(owner)) return keys; + var rows = owners.getAsJsonObject().get(owner); + if (!rows.isJsonObject()) throw new IllegalStateException("Unresolved cache owner structure"); + for (var entry : rows.getAsJsonObject().entrySet()) keys.add(entry.getKey()); + return keys; + } catch (IOException failure) { + throw new IllegalStateException("Unable to reserve persisted vote journal keys", failure); + } + } + protected boolean verifyJsonServerVote(String server, int index, OfflineBungeeVote vote) { try { return VoteCacheDurability.saveAndVerifyServerVote(jsonStorage, server, index, vote); diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java index a59d9c451..adb6fc153 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java @@ -67,7 +67,8 @@ public synchronized void reload() { var loaded = GsonConfigurationLoader.builder() .source(() -> new java.io.BufferedReader(new java.io.StringReader(document.toString()))) .build().load(); - if (!loaded.isMap()) throw new IOException("Vote cache must contain a JSON object"); + // readJournal already verifies the raw object. Configurate can normalize + // an object containing only null cache entries to an empty root. super.setConf(loaded); writesPaused = false; journalLoaded = true; diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java index fd198ec4a..0dcffe01b 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java @@ -892,6 +892,104 @@ Stream healthyUpdatesAndRemovalsPreserveMalformedRowsAndRestartFenc }); } + @TestFactory + Stream sqlInsertFallbackReservesRawNullKeysAndRetainsNewVoteIdentity() { + return lanes((velocity, online) -> { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject rows = new JsonObject(), owners = new JsonObject(); + rows.add("0", com.google.gson.JsonNull.INSTANCE); + owners.add(owner(online), rows); + journal.add(online ? "OnlineCache" : "VoteCache", owners); + Files.writeString(prepared.storage().getStoragePath(), journal.toString()); + IVoteCache storage = cache(velocity); + VoteCacheHandler recovering = sqlHandler(storage, online, List.of(), List.of()); + recovering.load(); + OfflineBungeeVote fresh = vote(UUID.randomUUID()); + // The SQL mock returns zero for insertion, exercising the actual emergency fallback. + add(recovering, online, fresh); + assertEquals("1", online ? fresh.getOnlineVoteCacheJsonKey() : fresh.getServerVoteCacheJsonKey()); + JsonObject saved = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertTrue(journalRows(saved, online).has("0")); + assertTrue(journalRows(saved, online).get("0").isJsonNull()); + assertEquals(fresh.getVoteId().toString(), journalRows(saved, online).getAsJsonObject("1").get("VoteId").getAsString()); + assertFalse(recovering.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + assertFalse(fresh.hasCachedVoteBatch()); + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, List.of(), List.of()); + restarted.load(); + assertEquals(1, votes(restarted, online).size()); + OfflineBungeeVote restored = votes(restarted, online).get(0); + assertEquals(fresh.getVoteId(), restored.getVoteId()); + assertEquals("1", online ? restored.getOnlineVoteCacheJsonKey() : restored.getServerVoteCacheJsonKey()); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(restored), 1)); + JsonObject reopened = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertEquals(journalRows(saved, online), journalRows(reopened, online)); + verifyOnlyCheckedSqlReads(restarted); + }); + } + + @TestFactory + Stream invalidBatchOwnerDeliveryUpdatesPersistWithoutReplacingReplayEvidence() { + return lanes((velocity, online) -> { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject malformed = sourceRow(journal, online, 41); + malformed.addProperty("CachedBatchNumber", "invalid-position"); + Files.writeString(prepared.storage().getStoragePath(), journal.toString()); + IVoteCache storage = cache(velocity); + VoteCacheHandler recovering = sqlHandler(storage, online, prepared.sqlVotes(), List.of(41, 42)); + recovering.load(); + OfflineBungeeVote invalid = sqlRow(recovering, online, 41), clean = sqlRow(recovering, online, 42); + assertEquals(prepared.identities().get(0), invalid.getVoteId()); + assertEquals(41, invalid.getCachedBatchSourceRowId()); + assertFalse(invalid.isCachedVoteBatchValid()); + assertFalse(invalid.isCachedBatchDurable()); + ProxyVoteCacheTable serverTable = (ProxyVoteCacheTable) getField(recovering, "voteCacheTable"); + ProxyOnlineVoteCacheTable onlineTable = (ProxyOnlineVoteCacheTable) getField(recovering, "onlineVoteCacheTable"); + if (online) when(onlineTable.updateProxyBroadcastState(any())).thenReturn(true); + else when(serverTable.updateProxyBroadcastState(any(), eq(BACKEND))).thenReturn(true); + invalid.setRewardDelivered(true); + assertTrue(online ? recovering.updateOnlineVote(PLAYER, invalid) : recovering.updateServerVote(BACKEND, invalid)); + // The next checked SQL snapshot reflects the successful SQL delivery-state update. + prepared.sqlVotes().get(0).setRewardDelivered(true); + JsonObject expected = malformed.deepCopy(); + expected.addProperty("RewardDelivered", true); + JsonObject updated = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertEquals(expected, sourceRow(updated, online, 41), "delivery success did not preserve updated flags and original evidence"); + assertFalse(recovering.prepareCachedVoteBatch(owner(online), online, List.of(invalid), 1)); + + clean.setRewardDelivered(true); + assertTrue(online ? recovering.updateOnlineVote(PLAYER, clean) : recovering.updateServerVote(BACKEND, clean)); + JsonObject healthyUpdated = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertTrue(sourceRow(healthyUpdated, online, 42).get("RewardDelivered").getAsBoolean()); + remove(recovering, online, clean); + JsonObject afterRemoval = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertEquals(expected, sourceRow(afterRemoval, online, 41)); + assertEquals(1, journalRows(afterRemoval, online).size()); + + // A different occurrence written at the occupied key cannot claim its delivery state or evidence. + String occupiedKey = journalRows(afterRemoval, online).keySet().iterator().next(); + OfflineBungeeVote differentIdentity = vote(UUID.randomUUID()); + differentIdentity.setRewardDelivered(false); + if (online) storage.addVoteOnline(PLAYER, Integer.parseInt(occupiedKey), differentIdentity); + else storage.addVote(BACKEND, Integer.parseInt(occupiedKey), differentIdentity); + storage.saveDurably(); + JsonObject guarded = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertEquals(expected, journalRows(guarded, online).getAsJsonObject(occupiedKey)); + VoteCacheHandler restarted = sqlHandler(cache(velocity), online, List.of(prepared.sqlVotes().get(0)), List.of(41)); + restarted.load(); + assertEquals(1, votes(restarted, online).size()); + OfflineBungeeVote restored = sqlRow(restarted, online, 41); + assertEquals(invalid.getVoteId(), restored.getVoteId()); + assertTrue(restored.isRewardDelivered()); + assertFalse(restored.isCachedVoteBatchValid()); + assertFalse(restored.isCachedBatchDurable()); + assertFalse(restarted.prepareCachedVoteBatch(owner(online), online, List.of(restored), 1)); + assertEquals(expected, sourceRow(JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(), online, 41)); + verifyOnlyCheckedSqlReads(restarted); + }); + } + @TestFactory Stream unusedNullFieldsAndIgnoredVoteIdAliasStillAllowLegacyBatchStaging() { return lanes((velocity, online) -> { diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VoteCacheHandlerVoteIdTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VoteCacheHandlerVoteIdTest.java index 06bdab548..597b07719 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VoteCacheHandlerVoteIdTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/VoteCacheHandlerVoteIdTest.java @@ -77,8 +77,9 @@ public void legacySqlRowConstructorsPreserveUnknownDelayValidation() { private VoteCacheHandler handler; @BeforeEach - public void setUp() { + public void setUp() throws Exception { storage = mock(IVoteCache.class); + Files.writeString(tempDir.resolve("vote-cache.json"), "{}"); when(storage.getStoragePath()).thenReturn(tempDir.resolve("vote-cache.json")); when(storage.getServerVotes("server")).thenReturn(Collections.emptyList()); when(storage.getOnlineVotes("player-uuid")).thenReturn(Collections.emptyList()); @@ -296,6 +297,7 @@ public void failedExpiredOnlineVoteRemovalKeepsInMemoryEntries() { public void failedDurableServerSaveRollsBackJsonMutationBeforeRetry() throws Exception { Path journal = Files.createTempFile("votingplugin-server-rollback", ".json"); try { + Files.writeString(journal, "{}"); ArrayList keys = new ArrayList<>(); when(storage.getStoragePath()).thenReturn(journal); when(storage.getServerVotes("server")).thenAnswer(invocation -> new ArrayList<>(keys)); @@ -325,6 +327,7 @@ public void failedDurableServerSaveRollsBackJsonMutationBeforeRetry() throws Exc public void failedDurableOnlineSaveRollsBackJsonMutationBeforeRetry() throws Exception { Path journal = Files.createTempFile("votingplugin-online-rollback", ".json"); try { + Files.writeString(journal, "{}"); ArrayList keys = new ArrayList<>(); when(storage.getStoragePath()).thenReturn(journal); when(storage.getOnlineVotes("player-uuid")).thenAnswer(invocation -> new ArrayList<>(keys)); @@ -354,6 +357,7 @@ public void failedDurableOnlineSaveRollsBackJsonMutationBeforeRetry() throws Exc public void failedRollbackReloadQuarantinesJsonJournal() throws Exception { Path journal = Files.createTempFile("votingplugin-quarantined-journal", ".json"); try { + Files.writeString(journal, "{}"); when(storage.getStoragePath()).thenReturn(journal); doThrow(new java.io.IOException("save failed")).when(storage).saveDurably(); doThrow(new IllegalStateException("reload failed")).when(storage).reload(); @@ -382,6 +386,7 @@ public void failedRollbackReloadQuarantinesJsonJournal() throws Exception { public void serverVoteIsNotExposedWhenJsonSaveDoesNotReachDisk() throws Exception { Path journal = Files.createTempFile("votingplugin-empty-journal", ".json"); try { + Files.writeString(journal, "{}"); when(storage.getStoragePath()).thenReturn(journal); handler = newVerifyingHandler(storage); @@ -397,6 +402,7 @@ public void serverVoteIsNotExposedWhenJsonSaveDoesNotReachDisk() throws Exceptio public void serverVoteUpdateIsRejectedWhenJsonSaveDoesNotReachDisk() throws Exception { Path journal = Files.createTempFile("votingplugin-empty-update-journal", ".json"); try { + Files.writeString(journal, "{}"); DataNode voteNode = mock(DataNode.class); when(voteNode.isObject()).thenReturn(true); stubString(voteNode, "UUID", "player-uuid"); From c7f5b786bb400c5b3e6f5fdf75724a95ed46d130 Mon Sep 17 00:00:00 2001 From: Ben Date: Tue, 13 Oct 2026 15:30:58 -0600 Subject: [PATCH 3/4] Preserve opaque recovery rows and quarantine malformed timed entries --- .../proxy/cache/VoteCacheDurability.java | 13 +- .../proxy/cache/VoteCacheHandler.java | 16 ++- .../proxy/velocity/VelocityJsonVoteCache.java | 11 ++ .../tests/CachedVoteBatchDurabilityTest.java | 129 ++++++++++++++++++ 4 files changed, 167 insertions(+), 2 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java index 1c7506020..269a4a54e 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java @@ -76,7 +76,7 @@ private static JsonObject objectChild(JsonObject parent, String key) { private static boolean unresolvedRow(JsonElement row) { if (!row.isJsonObject()) return true; JsonObject data = row.getAsJsonObject(); - if (hasUndecodableVoteFields(data)) return true; + if (isOpaqueRecoveryRow(new GsonDataNode(data)) || hasUndecodableVoteFields(data)) return true; if (!data.has("CachedBatchNumber") && !data.has("CachedBatchSize") && !data.has("CachedBatchSourceRowId")) return false; try { @@ -100,6 +100,17 @@ private static boolean unresolvedRow(JsonElement row) { } catch (RuntimeException invalid) { return true; } } + /** Shares opaque-row ownership classification between reconstruction and persistence. */ + static boolean isOpaqueRecoveryRow(DataNode data) { + for (String field : new String[] {"Name", "UUID", "Service", "Time", "Real", "Text", "VoteId", "VoteID"}) { + if (data.has(field)) { + DataNode value = data.get(field); + if (value != null && value.isPrimitive()) return false; + } + } + return true; + } + private static boolean hasUndecodableVoteFields(JsonObject data) { // Match fields which recovery actually reads, including optional delivery state. // Do not classify unused/unknown null fields as broken legacy ownership. diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java index b30e4d64f..3b27c3093 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheHandler.java @@ -1843,8 +1843,20 @@ private boolean sqlJournalStructureIsValid(com.google.gson.JsonObject document) if (!owners.isJsonObject()) return false; for (var owner : owners.getAsJsonObject().entrySet()) { if (!owner.getValue().isJsonObject()) return false; - for (var row : owner.getValue().getAsJsonObject().entrySet()) + Collection loadedKeys = lane.equals("VoteCache") ? jsonStorage.getServerVotes(owner.getKey()) + : jsonStorage.getOnlineVotes(owner.getKey()); + for (var row : owner.getValue().getAsJsonObject().entrySet()) { if (!row.getValue().isJsonObject()) return false; + // A normalized-away binding is unresolved even when its raw object + // survives structural parsing. Inspect every raw row before clearing. + if (loadedKeys == null || !loadedKeys.contains(row.getKey())) return false; + try { + DataNode data = new GsonDataNode(row.getValue()); + validateSqlJournalIdentity(data); + OfflineBungeeVote vote = decodeJsonVote(data); + if (vote == null || !vote.isCachedVoteBatchValid()) return false; + } catch (RuntimeException invalid) { return false; } + } } } return true; @@ -1859,6 +1871,8 @@ private DataNode persistedSqlJournalVote(com.google.gson.JsonObject document, St private void validateSqlJournalIdentity(DataNode data) { if (data == null || !data.isObject()) throw new IllegalArgumentException("Invalid SQL vote journal row"); + if (VoteCacheDurability.isOpaqueRecoveryRow(data)) + throw new IllegalArgumentException("Opaque SQL vote journal row"); // Old emergency rows without replay metadata retain their established optional fields. if (!data.has("CachedBatchNumber") && !data.has("CachedBatchSize") && !data.has("CachedBatchSourceRowId")) return; diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java index adb6fc153..90641fa25 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java @@ -67,6 +67,17 @@ public synchronized void reload() { var loaded = GsonConfigurationLoader.builder() .source(() -> new java.io.BufferedReader(new java.io.StringReader(document.toString()))) .build().load(); + // Timed outbox rows have no opaque-row preservation during save. Never + // accept a raw timed entry that Configurate would drop or reclassify. + if (document.has("TimedVoteCache")) { + var timed = document.get("TimedVoteCache"); + if (!timed.isJsonObject()) throw new IOException("Invalid timed vote cache structure"); + for (var entry : timed.getAsJsonObject().entrySet()) { + if (!entry.getValue().isJsonObject() || (!loaded.node("TimedVoteCache", entry.getKey()).isMap() + || loaded.node("TimedVoteCache", entry.getKey()).childrenMap().isEmpty())) + throw new IOException("Timed vote cache entry requires recovery"); + } + } // readJournal already verifies the raw object. Configurate can normalize // an object containing only null cache entries to an empty root. super.setConf(loaded); diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java index 0dcffe01b..8789519d4 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java @@ -990,6 +990,135 @@ Stream invalidBatchOwnerDeliveryUpdatesPersistWithoutReplacingRepla }); } + @TestFactory + Stream rawOpaqueOrdinaryRowsFenceBothLanesUntilCompleteRepair() { + return lanes((velocity, online) -> { + for (String raw : List.of("{\"CachedBatchNumber\":null,\"CachedBatchSize\":null,\"CachedBatchSourceRowId\":null}", + "{}", "{\"Name\":null,\"UUID\":null,\"Service\":null,\"Time\":null,\"Real\":null,\"Text\":null}", + "{\"VoteId\":null}", "{\"HttpDeliveryIds\":null}", "{\"UnusedOptionalField\":null}", + "{\"CachedBatchNumber\":{},\"CachedBatchSize\":{},\"CachedBatchSourceRowId\":{}}", + "{\"UnusedOptionalField\":\"Name = not a vote; UUID = prose\"}", + "{\"UnusedOptionalField\":{}}", "{\"UnusedOptionalField\":[]}")) { + PreparedJournal prepared = preparedJournal(velocity, online, false); + JsonObject journal = JsonParser.parseString(prepared.cleanBytes()).getAsJsonObject(); + JsonObject rows = new JsonObject(), owners = new JsonObject(); + rows.add("9", JsonParser.parseString(raw)); owners.add(owner(!online), rows); + String lane = online ? "VoteCache" : "OnlineCache"; + journal.add(lane, owners); + String malformed = journal.toString(); + Files.writeString(prepared.storage().getStoragePath(), malformed); + IVoteCache storage = cache(velocity); + VoteCacheHandler recovering = sqlHandler(storage, online, + List.of(prepared.sqlVotes().get(0), prepared.sqlVotes().get(1), vote(null)), List.of(41, 42, 43)); + recovering.load(); + OfflineBungeeVote clean = sqlRow(recovering, online, 42), fresh = sqlRow(recovering, online, 43); + assertTrue(clean.isCachedBatchDurable()); + assertTrue(recovering.prepareCachedVoteBatch(owner(online), online, List.of(clean), 1)); + assertFalse(recovering.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1), raw); + assertNull(fresh.getVoteId()); assertFalse(fresh.hasCachedVoteBatch()); + assertEquals(malformed, Files.readString(storage.getStoragePath())); + storage.saveDurably(); + JsonObject saved = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertNotNull(saved.get(lane), "opaque lane disappeared"); + assertNotNull(saved.getAsJsonObject(lane).get(owner(!online)), "opaque owner disappeared"); + assertEquals(JsonParser.parseString(raw), journalRows(saved, !online).get("9"), "routine save erased opaque row"); + ProxyVoteCacheTable serverTable = (ProxyVoteCacheTable) getField(recovering, "voteCacheTable"); + ProxyOnlineVoteCacheTable onlineTable = (ProxyOnlineVoteCacheTable) getField(recovering, "onlineVoteCacheTable"); + if (online) when(onlineTable.updateProxyBroadcastState(any())).thenReturn(true); + else when(serverTable.updateProxyBroadcastState(any(), eq(BACKEND))).thenReturn(true); + clean.setRewardDelivered(true); + assertTrue(online ? recovering.updateOnlineVote(PLAYER, clean) : recovering.updateServerVote(BACKEND, clean)); + prepared.sqlVotes().get(1).setRewardDelivered(true); + if (online) verify(onlineTable).updateProxyBroadcastState(clean); + else verify(serverTable).updateProxyBroadcastState(clean, BACKEND); + saved = JsonParser.parseString(Files.readString(storage.getStoragePath())).getAsJsonObject(); + assertEquals(JsonParser.parseString(raw), journalRows(saved, !online).get("9"), "healthy update erased opaque row"); + recovering.load(); + assertTrue(votes(recovering, !online).isEmpty(), "opaque prose became a cached reward owner"); + assertFalse(recovering.prepareCachedVoteBatch(owner(online), online, + List.of(sqlRow(recovering, online, 43)), 1), "save/reload cleared opaque ownership without repair"); + journal = saved; journal.remove(lane); Files.writeString(storage.getStoragePath(), journal.toString()); + storage.reload(); + assertFalse(recovering.prepareCachedVoteBatch(owner(online), online, List.of(fresh), 1)); + recovering.load(); + verifyOnlyCheckedSqlReads(recovering, 3); + assertTrue(recovering.prepareCachedVoteBatch(owner(online), online, List.of(sqlRow(recovering, online, 43)), 1)); + } + }); + } + + @TestFactory + Stream velocityMalformedTimedRecordsQuarantineWholeJournalWithoutLosingOwners() { + return Stream.of(false, true).flatMap(adjacent -> Stream.of("\n", "\r\n").flatMap(separator -> + Stream.of("null-lane", "null-row", "empty-row", "null-only-row", "scalar-row", "array-row") + .map(shape -> DynamicTest.dynamicTest(shape + "/adjacent=" + adjacent + "/CRLF=" + separator.equals("\r\n"), () -> { + Path factoryDirectory = directory; + directory = Files.createTempDirectory(factoryDirectory, "timed-boundary-"); + try { + IVoteCache storage = cache(true); + var timedVote = new com.bencodez.votingplugin.timequeue.VoteTimeQueue( + UUID.randomUUID(), "tést-player", "service-雪", 100L); + storage.addTimedVote(1, timedVote); + storage.addVote(BACKEND, 2, vote(UUID.randomUUID())); + storage.saveDurably(); + byte[] healthy = Files.readAllBytes(storage.getStoragePath()); + VoteCacheHandler recovering = handler(storage); + recovering.load(); + var ownedTimedVotes = recovering.getTimeChangeQueue(); + assertEquals(1, ownedTimedVotes.size()); + var owned = ownedTimedVotes.peek(); + JsonObject document = adjacent ? JsonParser.parseString(new String(healthy, + java.nio.charset.StandardCharsets.UTF_8)).getAsJsonObject() : new JsonObject(); + JsonObject rows = new JsonObject(); + if (adjacent) rows = document.getAsJsonObject("TimedVoteCache"); + switch (shape) { + case "null-row" -> rows.add("0", com.google.gson.JsonNull.INSTANCE); + case "empty-row" -> rows.add("0", new JsonObject()); + case "null-only-row" -> { + JsonObject row = new JsonObject(); row.add("Name", com.google.gson.JsonNull.INSTANCE); rows.add("0", row); + } + case "scalar-row" -> rows.addProperty("0", "TimedVoteCache is data; never execute this text"); + case "array-row" -> rows.add("0", new com.google.gson.JsonArray()); + default -> { } + } + document.add("TimedVoteCache", shape.equals("null-lane") ? com.google.gson.JsonNull.INSTANCE : rows); + // Exercise the complete persisted wrapper, UTF-8 data and physical line endings. + if (adjacent) document.addProperty("BoundaryNote", "雪 before TimedVoteCache; {\"TimedVoteCache\":null} is text\nnext line"); + String payload = new com.google.gson.GsonBuilder().serializeNulls().setPrettyPrinting().create().toJson(document) + .replace("\n", separator); + byte[] malformed = payload.getBytes(java.nio.charset.StandardCharsets.UTF_8); + Files.write(storage.getStoragePath(), malformed); + assertThrows(RuntimeException.class, recovering::load); + assertSame(ownedTimedVotes, recovering.getTimeChangeQueue()); + assertSame(owned, recovering.getTimeChangeQueue().peek()); + storage.save(); + assertThrows(IOException.class, storage::saveDurably); + assertArrayEquals(malformed, Files.readAllBytes(storage.getStoragePath())); + Files.write(storage.getStoragePath(), healthy); + recovering.load(); + assertEquals(1, recovering.getTimeChangeQueue().size()); + assertEquals(timedVote.getVoteId(), recovering.getTimeChangeQueue().peek().getVoteId()); + storage.saveDurably(); + assertEquals(timedVote.getVoteId().toString(), JsonParser.parseString(Files.readString(storage.getStoragePath())) + .getAsJsonObject().getAsJsonObject("TimedVoteCache").getAsJsonObject("1").get("VoteId").getAsString()); + } finally { directory = factoryDirectory; } + })))); + } + + @Test + void velocityEmptyTimedLaneAndCodeLikeAdjacentDataRemainValid() throws Exception { + IVoteCache storage = cache(true); + String note = "雪: {\"TimedVoteCache\":null}\nString codeLike = \"TimedVoteCache\";"; + JsonObject document = new JsonObject(); + document.add("TimedVoteCache", new JsonObject()); + document.addProperty("BoundaryNote", note); + Files.writeString(storage.getStoragePath(), document.toString()); + storage.reload(); storage.saveDurably(); + assertEquals(note, JsonParser.parseString(Files.readString(storage.getStoragePath())) + .getAsJsonObject().get("BoundaryNote").getAsString()); + assertTrue(storage.getTimedVoteCache().isEmpty()); + } + @TestFactory Stream unusedNullFieldsAndIgnoredVoteIdAliasStillAllowLegacyBatchStaging() { return lanes((velocity, online) -> { From 2dc99a0b9787b204107b21bc7ce9594d6cdaac2c Mon Sep 17 00:00:00 2001 From: Ben Date: Tue, 13 Oct 2026 16:04:03 -0600 Subject: [PATCH 4/4] Validate timed recovery rows before cache rewrites --- .../proxy/cache/VoteCacheDurability.java | 50 ++++- .../proxy/velocity/VelocityJsonVoteCache.java | 35 ++-- .../tests/CachedVoteBatchDurabilityTest.java | 174 ++++++++++++++++-- 3 files changed, 229 insertions(+), 30 deletions(-) diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java index 269a4a54e..9cdbe8fb4 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/cache/VoteCacheDurability.java @@ -41,7 +41,55 @@ public static JsonObject readJournal(Path path) throws IOException { JsonElement document = readValue(reader, 0); if (!document.isJsonObject() || reader.peek() != JsonToken.END_DOCUMENT) throw new IOException("Vote cache must contain one JSON object"); - return document.getAsJsonObject(); + JsonObject journal = document.getAsJsonObject(); + validateTimedRows(journal); + return journal; + } + } + + /** Rejects timed recovery evidence that queue reconstruction cannot safely retain. */ + private static void validateTimedRows(JsonObject journal) throws IOException { + if (!journal.has("TimedVoteCache")) return; + JsonElement timed = journal.get("TimedVoteCache"); + if (!timed.isJsonObject()) throw new IOException("Invalid timed vote cache structure"); + for (var entry : timed.getAsJsonObject().entrySet()) { + // Queue rewrites address rows by integer index. Aliases or out-of-range + // keys cannot retain distinct durable owners through that representation. + if (!entry.getKey().matches("0|[1-9][0-9]*")) + throw new IOException("Timed vote cache key requires recovery"); + try { Integer.parseInt(entry.getKey()); } + catch (NumberFormatException invalid) { throw new IOException("Timed vote cache key requires recovery"); } + JsonElement row = entry.getValue(); + if (!row.isJsonObject()) throw new IOException("Timed vote cache entry requires recovery"); + JsonObject fields = row.getAsJsonObject(); + boolean readable = false; + for (String name : new String[] {"Name", "UUID", "Service", "Time", "VoteId"}) + if (fields.has(name) && fields.get(name).isJsonPrimitive()) readable = true; + if (!readable) throw new IOException("Timed vote cache entry requires recovery"); + try { + // Check the fields consumed by timed reconstruction, including optional + // progress. Unknown values and inactive conditional fields remain legacy data. + for (String name : new String[] {"Name", "UUID", "Service", "Time", "VoteId", + "ProxyBroadcastHandled", "BroadcastForwardedServers", "BroadcastTargets", "Totals", + "Processed", "VotePartyApplied", "TotalsApplied", "MultiProxyForwardingHandled", + "HttpBroadcastDeliveryIds", "MultiProxyForwardingRequired", "MultiProxyCapabilityDiscoveryPending", + "RealVote", "WasOnlineKnown", "DelayValidationKnown", "MultiProxyOrigin", + "MultiProxyCompletionPending", "MultiProxyRecipients", "MultiProxyAcknowledgedServers", + "MultiProxyLegacyPendingRecipients"}) + if (fields.has(name) && !fields.get(name).isJsonPrimitive()) + throw new IOException("Timed vote cache entry requires recovery"); + if (fields.has("Time")) fields.get("Time").getAsLong(); + if (fields.has("VoteId") && !fields.get("VoteId").getAsString().isEmpty()) + java.util.UUID.fromString(fields.get("VoteId").getAsString()); + if (fields.has("WasOnlineKnown") && fields.get("WasOnlineKnown").getAsBoolean() + && (!fields.has("WasOnline") || !fields.get("WasOnline").isJsonPrimitive())) + throw new IOException("Timed vote cache entry requires recovery"); + if (fields.has("DelayValidationKnown") && fields.get("DelayValidationKnown").getAsBoolean() + && fields.has("DelayValidated") && !fields.get("DelayValidated").isJsonPrimitive()) + throw new IOException("Timed vote cache entry requires recovery"); + } catch (RuntimeException invalid) { + throw new IOException("Timed vote cache entry requires recovery"); + } } } diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java index 90641fa25..e5c4f59c7 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/velocity/VelocityJsonVoteCache.java @@ -67,17 +67,7 @@ public synchronized void reload() { var loaded = GsonConfigurationLoader.builder() .source(() -> new java.io.BufferedReader(new java.io.StringReader(document.toString()))) .build().load(); - // Timed outbox rows have no opaque-row preservation during save. Never - // accept a raw timed entry that Configurate would drop or reclassify. - if (document.has("TimedVoteCache")) { - var timed = document.get("TimedVoteCache"); - if (!timed.isJsonObject()) throw new IOException("Invalid timed vote cache structure"); - for (var entry : timed.getAsJsonObject().entrySet()) { - if (!entry.getValue().isJsonObject() || (!loaded.node("TimedVoteCache", entry.getKey()).isMap() - || loaded.node("TimedVoteCache", entry.getKey()).childrenMap().isEmpty())) - throw new IOException("Timed vote cache entry requires recovery"); - } - } + validateTimedRows(document, loaded); // readJournal already verifies the raw object. Configurate can normalize // an object containing only null cache entries to an empty root. super.setConf(loaded); @@ -92,6 +82,24 @@ public synchronized void reload() { } } + /** Checks raw recovery fields using the same normalized flag coercion as the timed decoder. */ + private static void validateTimedRows(JsonObject document, org.spongepowered.configurate.ConfigurationNode loaded) + throws IOException { + if (!document.has("TimedVoteCache")) return; + for (var entry : document.getAsJsonObject("TimedVoteCache").entrySet()) { + var node = loaded.node("TimedVoteCache", entry.getKey()); + if (!node.isMap() || node.childrenMap().isEmpty()) + throw new IOException("Timed vote cache entry requires recovery"); + JsonObject raw = entry.getValue().getAsJsonObject(); + if (node.node("WasOnlineKnown").getBoolean() + && (!raw.has("WasOnline") || !raw.get("WasOnline").isJsonPrimitive())) + throw new IOException("Timed vote cache entry requires recovery"); + if (node.node("DelayValidationKnown").getBoolean() + && raw.has("DelayValidated") && !raw.get("DelayValidated").isJsonPrimitive()) + throw new IOException("Timed vote cache entry requires recovery"); + } + } + @Override public synchronized org.spongepowered.configurate.ConfigurationNode getData() { return super.getData(); } @Override @@ -113,6 +121,11 @@ public synchronized void saveDurably() throws IOException { if (journalLoaded) { try { previous = VoteCacheDurability.readJournal(target); + String rawPrevious = previous.toString(); + var previousNodes = GsonConfigurationLoader.builder() + .source(() -> new java.io.BufferedReader(new java.io.StringReader(rawPrevious))) + .build().load(); + validateTimedRows(previous, previousNodes); PendingPluginMessageVotePartyRewards.fromCacheJson(previous); } catch (IOException | RuntimeException failure) { diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java index 8789519d4..e1cf386a3 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/tests/CachedVoteBatchDurabilityTest.java @@ -1048,14 +1048,16 @@ Stream rawOpaqueOrdinaryRowsFenceBothLanesUntilCompleteRepair() { } @TestFactory - Stream velocityMalformedTimedRecordsQuarantineWholeJournalWithoutLosingOwners() { - return Stream.of(false, true).flatMap(adjacent -> Stream.of("\n", "\r\n").flatMap(separator -> - Stream.of("null-lane", "null-row", "empty-row", "null-only-row", "scalar-row", "array-row") - .map(shape -> DynamicTest.dynamicTest(shape + "/adjacent=" + adjacent + "/CRLF=" + separator.equals("\r\n"), () -> { + Stream malformedTimedRecordsQuarantineWholeJournalWithoutLosingOwners() { + return Stream.of(false, true).flatMap(velocity -> Stream.of(false, true).flatMap(adjacent -> Stream.of("\n", "\r\n").flatMap(separator -> + Stream.of("null-lane", "null-row", "empty-row", "null-only-row", "scalar-row", "array-row", + "unknown-null-row", "unknown-prose-row", "nested-row", "array-field-row", + "aliased-key", "overflow-key", "negative-key", "text-key", "unicode-key") + .map(shape -> DynamicTest.dynamicTest((velocity ? "Velocity/" : "Bungee/") + shape + "/adjacent=" + adjacent + "/CRLF=" + separator.equals("\r\n"), () -> { Path factoryDirectory = directory; directory = Files.createTempDirectory(factoryDirectory, "timed-boundary-"); try { - IVoteCache storage = cache(true); + IVoteCache storage = cache(velocity); var timedVote = new com.bencodez.votingplugin.timequeue.VoteTimeQueue( UUID.randomUUID(), "tést-player", "service-雪", 100L); storage.addTimedVote(1, timedVote); @@ -1079,6 +1081,25 @@ Stream velocityMalformedTimedRecordsQuarantineWholeJournalWithoutLo } case "scalar-row" -> rows.addProperty("0", "TimedVoteCache is data; never execute this text"); case "array-row" -> rows.add("0", new com.google.gson.JsonArray()); + case "unknown-null-row", "unknown-prose-row", "nested-row", "array-field-row" -> { + JsonObject row = new JsonObject(); + if (shape.equals("unknown-null-row")) row.add("Note", com.google.gson.JsonNull.INSTANCE); + else if (shape.equals("unknown-prose-row")) row.addProperty("Note", "Name and VoteId are prose, not fields"); + else row.add("Name", shape.equals("nested-row") ? new JsonObject() : new com.google.gson.JsonArray()); + rows.add("0", row); + } + case "aliased-key", "overflow-key", "negative-key", "text-key", "unicode-key" -> { + String key = switch (shape) { + case "aliased-key" -> "01"; + case "overflow-key" -> "2147483648"; + case "negative-key" -> "-1"; + case "unicode-key" -> "１"; + default -> "owner"; + }; + JsonObject row = new JsonObject(); row.addProperty("Name", "synthetic-player"); + row.addProperty("Service", "synthetic-service"); row.addProperty("Time", 100L); + row.addProperty("VoteId", UUID.randomUUID().toString()); rows.add(key, row); + } default -> { } } document.add("TimedVoteCache", shape.equals("null-lane") ? com.google.gson.JsonNull.INSTANCE : rows); @@ -1088,6 +1109,11 @@ Stream velocityMalformedTimedRecordsQuarantineWholeJournalWithoutLo .replace("\n", separator); byte[] malformed = payload.getBytes(java.nio.charset.StandardCharsets.UTF_8); Files.write(storage.getStoragePath(), malformed); + // Completion must not rewrite away a malformed sibling, even before reload. + owned.setProcessed(true); + assertFalse(recovering.removeTimeVote(owned)); + assertSame(owned, recovering.getTimeChangeQueue().peek()); + assertArrayEquals(malformed, Files.readAllBytes(storage.getStoragePath())); assertThrows(RuntimeException.class, recovering::load); assertSame(ownedTimedVotes, recovering.getTimeChangeQueue()); assertSame(owned, recovering.getTimeChangeQueue().peek()); @@ -1101,22 +1127,134 @@ Stream velocityMalformedTimedRecordsQuarantineWholeJournalWithoutLo storage.saveDurably(); assertEquals(timedVote.getVoteId().toString(), JsonParser.parseString(Files.readString(storage.getStoragePath())) .getAsJsonObject().getAsJsonObject("TimedVoteCache").getAsJsonObject("1").get("VoteId").getAsString()); + var repaired = recovering.getTimeChangeQueue().peek(); + repaired.setProcessed(true); + assertTrue(recovering.removeTimeVote(repaired)); + assertTrue(recovering.getTimeChangeQueue().isEmpty()); + assertTrue(storage.getTimedVoteCache().isEmpty()); } finally { directory = factoryDirectory; } - })))); + }))))); } - @Test - void velocityEmptyTimedLaneAndCodeLikeAdjacentDataRemainValid() throws Exception { - IVoteCache storage = cache(true); - String note = "雪: {\"TimedVoteCache\":null}\nString codeLike = \"TimedVoteCache\";"; - JsonObject document = new JsonObject(); - document.add("TimedVoteCache", new JsonObject()); - document.addProperty("BoundaryNote", note); - Files.writeString(storage.getStoragePath(), document.toString()); - storage.reload(); storage.saveDurably(); - assertEquals(note, JsonParser.parseString(Files.readString(storage.getStoragePath())) - .getAsJsonObject().get("BoundaryNote").getAsString()); - assertTrue(storage.getTimedVoteCache().isEmpty()); + @TestFactory + Stream malformedConsumedTimedFieldsCannotBeErasedByCompletion() { + return Stream.of(false, true).map(velocity -> DynamicTest.dynamicTest(velocity ? "Velocity" : "Bungee", () -> { + String[] fields = {"Service", "Name", "UUID", "Time", "VoteId", "ProxyBroadcastHandled", + "BroadcastForwardedServers", "BroadcastTargets", "Totals", "Processed", "VotePartyApplied", + "TotalsApplied", "MultiProxyForwardingHandled", "HttpBroadcastDeliveryIds", "MultiProxyForwardingRequired", + "MultiProxyCapabilityDiscoveryPending", "RealVote", "WasOnlineKnown", "WasOnline", + "DelayValidationKnown", "DelayValidated", "MultiProxyOrigin", "MultiProxyCompletionPending", + "MultiProxyRecipients", "MultiProxyAcknowledgedServers", "MultiProxyLegacyPendingRecipients"}; + Path factoryDirectory = directory; + try { + for (String field : fields) for (int shape = 0; shape < 5; shape++) { + if (shape == 3 && !field.equals("Time") && !field.equals("VoteId")) continue; + if (shape == 4 && !field.equals("WasOnline")) continue; + directory = Files.createTempDirectory(factoryDirectory, "timed-field-boundary-"); + IVoteCache storage = cache(velocity); + var vote = new com.bencodez.votingplugin.timequeue.VoteTimeQueue(UUID.randomUUID(), "synthetic-player", "service-雪", 100L); + storage.addTimedVote(1, vote); storage.saveDurably(); + byte[] healthy = Files.readAllBytes(storage.getStoragePath()); + VoteCacheHandler recovering = handler(storage); recovering.load(); + var ownedQueue = recovering.getTimeChangeQueue(); var owned = ownedQueue.peek(); + JsonObject document = JsonParser.parseString(new String(healthy, java.nio.charset.StandardCharsets.UTF_8)).getAsJsonObject(); + JsonObject malformed = document.getAsJsonObject("TimedVoteCache").getAsJsonObject("1").deepCopy(); + malformed.addProperty("VoteId", UUID.randomUUID().toString()); + malformed.addProperty("WasOnlineKnown", true); malformed.addProperty("WasOnline", true); + malformed.addProperty("DelayValidationKnown", true); malformed.addProperty("DelayValidated", true); + if (shape == 4) malformed.remove(field); + else malformed.add(field, shape == 0 ? new JsonObject() : shape == 1 ? new com.google.gson.JsonArray() + : shape == 2 ? com.google.gson.JsonNull.INSTANCE : new com.google.gson.JsonPrimitive("not-a-valid-value")); + document.getAsJsonObject("TimedVoteCache").add("0", malformed); + document.addProperty("BoundaryNote", "雪: {\"Service\":{}} is data"); + byte[] broken = new com.google.gson.GsonBuilder().serializeNulls().setPrettyPrinting().create().toJson(document).getBytes(java.nio.charset.StandardCharsets.UTF_8); + Files.write(storage.getStoragePath(), broken); + owned.setProcessed(true); + assertFalse(recovering.removeTimeVote(owned), field + "/" + shape); + assertArrayEquals(broken, Files.readAllBytes(storage.getStoragePath())); + assertThrows(RuntimeException.class, recovering::load); + assertSame(ownedQueue, recovering.getTimeChangeQueue()); assertSame(owned, ownedQueue.peek()); + assertThrows(IOException.class, storage::saveDurably); + Files.write(storage.getStoragePath(), healthy); recovering.load(); + var repaired = recovering.getTimeChangeQueue().peek(); repaired.setProcessed(true); + assertTrue(recovering.removeTimeVote(repaired)); + assertTrue(recovering.getTimeChangeQueue().isEmpty()); + } + } finally { directory = factoryDirectory; } + })); + } + + @TestFactory + Stream velocityConditionalCoercionCannotHideMalformedTimedRecoveryFields() { + return Stream.of("WasOnline", "DelayValidated").map(field -> DynamicTest.dynamicTest(field, () -> { + Path factoryDirectory = directory; + try { + for (boolean numeric : new boolean[] {false, true}) for (int shape = 0; shape < 3; shape++) { + directory = Files.createTempDirectory(factoryDirectory, "timed-coercion-"); + IVoteCache storage = cache(true); + var vote = new com.bencodez.votingplugin.timequeue.VoteTimeQueue(UUID.randomUUID(), "synthetic-player", "service-雪", 100L); + storage.addTimedVote(1, vote); storage.saveDurably(); byte[] healthy = Files.readAllBytes(storage.getStoragePath()); + VoteCacheHandler recovering = handler(storage); recovering.load(); var ownedQueue = recovering.getTimeChangeQueue(); var owned = ownedQueue.peek(); + JsonObject document = JsonParser.parseString(new String(healthy, java.nio.charset.StandardCharsets.UTF_8)).getAsJsonObject(); + JsonObject bad = document.getAsJsonObject("TimedVoteCache").getAsJsonObject("1").deepCopy(); + bad.addProperty("VoteId", UUID.randomUUID().toString()); + String known = field.equals("WasOnline") ? "WasOnlineKnown" : "DelayValidationKnown"; + if (numeric) bad.addProperty(known, 1); else bad.addProperty(known, "yes"); + bad.add(field, shape == 0 ? new JsonObject() : shape == 1 ? new com.google.gson.JsonArray() : com.google.gson.JsonNull.INSTANCE); + document.getAsJsonObject("TimedVoteCache").add("0", bad); + byte[] broken = new com.google.gson.GsonBuilder().serializeNulls().create().toJson(document).getBytes(java.nio.charset.StandardCharsets.UTF_8); + Files.write(storage.getStoragePath(), broken); owned.setProcessed(true); + assertFalse(recovering.removeTimeVote(owned)); assertArrayEquals(broken, Files.readAllBytes(storage.getStoragePath())); + assertThrows(RuntimeException.class, recovering::load); assertSame(ownedQueue, recovering.getTimeChangeQueue()); assertSame(owned, ownedQueue.peek()); + assertThrows(IOException.class, storage::saveDurably); + Files.write(storage.getStoragePath(), healthy); recovering.load(); var repaired = recovering.getTimeChangeQueue().peek(); repaired.setProcessed(true); + assertTrue(recovering.removeTimeVote(repaired)); assertTrue(recovering.getTimeChangeQueue().isEmpty()); + } + } finally { directory = factoryDirectory; } + })); + } + + @TestFactory + Stream emptyTimedLaneAndLegacyRowsWithCodeLikeAdjacentDataRemainValid() { + return Stream.of(false, true).flatMap(velocity -> Stream.of(false, true).map(coerced -> DynamicTest.dynamicTest((velocity ? "Velocity" : "Bungee") + "/coerced=" + coerced, () -> { + Path factoryDirectory = directory; + directory = Files.createTempDirectory(factoryDirectory, "valid-timed-boundary-"); + try { + IVoteCache storage = cache(velocity); + String note = "雪: {\"TimedVoteCache\":null}\nString codeLike = \"TimedVoteCache\";"; + JsonObject document = new JsonObject(); + document.add("TimedVoteCache", new JsonObject()); + document.addProperty("BoundaryNote", note); + Files.writeString(storage.getStoragePath(), document.toString()); + storage.reload(); storage.saveDurably(); + assertEquals(note, JsonParser.parseString(Files.readString(storage.getStoragePath())) + .getAsJsonObject().get("BoundaryNote").getAsString()); + assertTrue(storage.getTimedVoteCache().isEmpty()); + JsonObject legacy = new JsonObject(); + legacy.addProperty("Name", "synthetic-player"); + legacy.addProperty("Service", "synthetic-service"); + legacy.addProperty("Time", 100L); + legacy.add("UnknownOptionalField", com.google.gson.JsonNull.INSTANCE); + legacy.addProperty("WasOnlineKnown", false); legacy.add("WasOnline", com.google.gson.JsonNull.INSTANCE); + legacy.addProperty("DelayValidationKnown", false); legacy.add("DelayValidated", com.google.gson.JsonNull.INSTANCE); + if (coerced) { + legacy.addProperty("WasOnlineKnown", "yes"); legacy.addProperty("WasOnline", true); + legacy.addProperty("DelayValidationKnown", 1); legacy.addProperty("DelayValidated", true); + } + document.getAsJsonObject("TimedVoteCache").add("3", legacy); + Files.writeString(storage.getStoragePath(), new com.google.gson.GsonBuilder().serializeNulls().create().toJson(document)); + VoteCacheHandler recovering = handler(storage); recovering.load(); + assertEquals(1, recovering.getTimeChangeQueue().size()); + var owned = recovering.getTimeChangeQueue().peek(); + assertEquals("synthetic-player", owned.getName()); + owned.setProcessed(true); + assertTrue(recovering.removeTimeVote(owned)); + assertTrue(recovering.getTimeChangeQueue().isEmpty()); + assertTrue(storage.getTimedVoteCache().isEmpty()); + assertEquals(note, JsonParser.parseString(Files.readString(storage.getStoragePath())) + .getAsJsonObject().get("BoundaryNote").getAsString()); + } finally { directory = factoryDirectory; } + }))); } @TestFactory