From be827b55d63e8c391e5d33343e447952ab103055 Mon Sep 17 00:00:00 2001 From: MadMax Date: Wed, 8 Jul 2026 16:00:26 +0100 Subject: [PATCH 1/3] feat(ahworker): prune applied journal rows --- cmake/MangosParams.cmake | 2 +- src/modules/AhWorker/Main.cpp | 237 ++++++++++++++++++- src/modules/AhWorker/ah-service.conf.dist.in | 40 +++- 3 files changed, 267 insertions(+), 12 deletions(-) diff --git a/cmake/MangosParams.cmake b/cmake/MangosParams.cmake index bccea9d02..92b697105 100644 --- a/cmake/MangosParams.cmake +++ b/cmake/MangosParams.cmake @@ -2,4 +2,4 @@ set(MANGOS_EXP "CLASSIC") set(MANGOS_PKG "Mangos Zero") set(MANGOS_WORLD_VER 2026070700) set(MANGOS_REALM_VER 2026060300) -set(MANGOS_AHBOT_VER 2026070700) +set(MANGOS_AHBOT_VER 2026070800) diff --git a/src/modules/AhWorker/Main.cpp b/src/modules/AhWorker/Main.cpp index ac6cd7f8f..f038ac582 100644 --- a/src/modules/AhWorker/Main.cpp +++ b/src/modules/AhWorker/Main.cpp @@ -78,11 +78,101 @@ #include "AuctionBook.h" #include "MutationHandler.h" +#include +#include +#include #include #include #include +#include +#include #include +static const uint32 AH_JOURNAL_APPLIED_RETENTION_SEC_DEFAULT = + 30u * 24u * 60u * 60u; +static const uint32 AH_JOURNAL_PRUNE_INTERVAL_SEC_DEFAULT = 60u * 60u; + +static bool AhParseNonNegativeSeconds(std::string const& text, uint32& result) +{ + if (text.empty()) + { + return false; + } + + uint64 value = 0u; + uint64 const maxValue = std::numeric_limits::max(); + for (std::string::const_iterator itr = text.begin(); itr != text.end(); ++itr) + { + if (*itr < '0' || *itr > '9') + { + return false; + } + + uint32 const digit = static_cast(*itr - '0'); + if (value > (maxValue - digit) / 10u) + { + return false; + } + value = value * 10u + digit; + } + + result = static_cast(value); + return true; +} + +static uint32 AhConfigNonNegativeSeconds(char const* name, uint32 defValue) +{ + char defText[16]; + snprintf(defText, sizeof(defText), "%u", defValue); + std::string const text = sConfig.GetStringDefault(name, defText); + + uint32 value = 0u; + if (AhParseNonNegativeSeconds(text, value)) + { + return value; + } + + fprintf(stderr, "ah-service: invalid %s value '%s' - using default %u\n", + name, text.c_str(), defValue); + return defValue; +} + +static bool AhJournalPruneDue(uint64 now, uint64& nextPrune, + uint32 intervalSec, uint32 retentionSec, + uint64& cutoff) +{ + if (now == std::numeric_limits::max()) + { + return false; + } + + if (intervalSec == 0u || retentionSec == 0u) + { + nextPrune = 0u; + return false; + } + + if (nextPrune == 0u) + { + nextPrune = now + intervalSec; + return false; + } + + if (now < nextPrune) + { + return false; + } + + nextPrune = now + intervalSec; + if (now <= retentionSec) + { + return false; + } + + cutoff = now - retentionSec; + return true; +} + // --------------------------------------------------------------------------- // Self-test: intent codec round-trip // --------------------------------------------------------------------------- @@ -1866,6 +1956,128 @@ static int RunJournalSelfTest() return 0; } +static int RunJournalPruneSchedulerSelfTest() +{ + char configPath[] = "ah-service-prune-selftest-XXXXXX"; + ACE_HANDLE const configHandle = ACE_OS::mkstemp(configPath); + if (configHandle == ACE_INVALID_HANDLE) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " could not create config fixture\n"); + return 1; + } + + FILE* configFile = ACE_OS::fdopen(configHandle, "w"); + if (configFile == nullptr) + { + ACE_OS::close(configHandle); + remove(configPath); + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " could not open config fixture\n"); + return 1; + } + + fprintf(configFile, + "[AhServiceConf]\n" + "AhTest.Valid = 3600\n" + "AhTest.Zero = 0\n" + "AhTest.Negative = -1\n" + "AhTest.Malformed = 3600oops\n" + "AhTest.Overflow = 4294967296\n"); + fclose(configFile); + + if (!sConfig.SetSource(configPath)) + { + remove(configPath); + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " could not load config fixture\n"); + return 1; + } + remove(configPath); + + uint32 const fallback = 77u; + if (AhConfigNonNegativeSeconds("AhTest.Valid", fallback) != 3600u || + AhConfigNonNegativeSeconds("AhTest.Zero", fallback) != 0u || + AhConfigNonNegativeSeconds("AhTest.Missing", fallback) != fallback || + AhConfigNonNegativeSeconds("AhTest.Negative", fallback) != fallback || + AhConfigNonNegativeSeconds("AhTest.Malformed", fallback) != fallback || + AhConfigNonNegativeSeconds("AhTest.Overflow", fallback) != fallback) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " unsafe config value did not use default\n"); + return 1; + } + + uint64 nextPrune = 0u; + uint64 cutoff = 0u; + + if (AhJournalPruneDue(1000u, nextPrune, 3600u, 100u, cutoff) || + nextPrune != 4600u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " first call should only schedule\n"); + return 1; + } + + if (AhJournalPruneDue(4599u, nextPrune, 3600u, 100u, cutoff)) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " early call pruned\n"); + return 1; + } + + if (!AhJournalPruneDue(4600u, nextPrune, 3600u, 100u, cutoff) || + cutoff != 4500u || nextPrune != 8200u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " due call did not prune with expected cutoff\n"); + return 1; + } + + nextPrune = 0u; + cutoff = 1234u; + if (AhJournalPruneDue(500u, nextPrune, 3600u, 1000u, cutoff) || + cutoff != 1234u || nextPrune != 4100u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " retention underflow should not prune\n"); + return 1; + } + + nextPrune = 777u; + cutoff = 1234u; + if (AhJournalPruneDue(5000u, nextPrune, 0u, 100u, cutoff) || + nextPrune != 0u || cutoff != 1234u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " zero interval should disable and reset\n"); + return 1; + } + + nextPrune = 777u; + if (AhJournalPruneDue(5000u, nextPrune, 3600u, 0u, cutoff) || + nextPrune != 0u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " zero retention should disable and reset\n"); + return 1; + } + + nextPrune = 777u; + cutoff = 1234u; + if (AhJournalPruneDue(std::numeric_limits::max(), nextPrune, + 3600u, 100u, cutoff) || + nextPrune != 777u || cutoff != 1234u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " failed wall clock should not prune\n"); + return 1; + } + + printf("journal prune scheduler selftest OK\n"); + return 0; +} + // --------------------------------------------------------------------------- // Self-test: SP-2 resolve outbox (worker expiry/win tick, Task 7) // --------------------------------------------------------------------------- @@ -3299,6 +3511,11 @@ int main(int argc, char** argv) { return rc; } + rc = RunJournalPruneSchedulerSelfTest(); + if (rc != 0) + { + return rc; + } rc = RunIntentCodecSelfTest(); if (rc != 0) { @@ -3619,6 +3836,13 @@ int main(int argc, char** argv) uint32 sinceMutTickMs = 0; const uint32 mutTickMs = static_cast( sConfig.GetIntDefault("AH.Service.TickMs", 1000)); + uint64 nextJournalPrune = 0u; + const uint32 journalPruneIntervalSec = AhConfigNonNegativeSeconds( + "AH.Service.JournalPruneIntervalSec", + AH_JOURNAL_PRUNE_INTERVAL_SEC_DEFAULT); + const uint32 journalAppliedRetentionSec = AhConfigNonNegativeSeconds( + "AH.Service.JournalAppliedRetentionSec", + AH_JOURNAL_APPLIED_RETENTION_SEC_DEFAULT); while (!stop) { @@ -3926,9 +4150,18 @@ int main(int argc, char** argv) // the dispatch above -- no tick-vs-handler race by construction). if (ahHandler != nullptr) { - ahHandler->CheckPrepareTimeouts(static_cast(time(NULL))); + uint64 const wallNow = static_cast(time(NULL)); + ahHandler->CheckPrepareTimeouts(wallNow); // SP-2 Task 8: re-send / abandon in-flight bot-sell materializations. - ahHandler->ResendStalePendingSells(static_cast(time(NULL))); + ahHandler->ResendStalePendingSells(wallNow); + + uint64 journalCutoff = 0u; + if (AhJournalPruneDue(wallNow, nextJournalPrune, + journalPruneIntervalSec, + journalAppliedRetentionSec, journalCutoff)) + { + AhJournal::DeleteAppliedOlderThan(botDb, journalCutoff); + } } // --- Bot cadence tick --- diff --git a/src/modules/AhWorker/ah-service.conf.dist.in b/src/modules/AhWorker/ah-service.conf.dist.in index 22225a2e2..68f28024e 100644 --- a/src/modules/AhWorker/ah-service.conf.dist.in +++ b/src/modules/AhWorker/ah-service.conf.dist.in @@ -38,18 +38,19 @@ WorldDatabaseConnections = 1 # CharacterDatabaseInfo # Character database connection string. Same host;port;user;pass;db # format as mangosd.conf. The ah-service child opens its OWN character-DB -# connection used by MarketSnapshot to read the live auction tables, and -# only ever issues SELECTs against it. +# connection. It reads the live auction tables in both modes. With +# WriteAuthority enabled it also writes the auction book and worker +# journal directly. # -# SECURITY: as with WorldDatabaseInfo, do NOT copy mangosd's full -# read/write credentials here. Use a DEDICATED least-privilege MySQL -# account with SELECT-only grants so a compromised child cannot write the -# DB directly and bypass the mangosd executor authority. See the -# "Security" section of doc/AuctionHouseBot.md. +# SECURITY: use a DEDICATED least-privilege MySQL account, not mangosd's +# full read/write credentials. Read-only service mode needs SELECT only. +# WriteAuthority additionally requires INSERT, UPDATE, and DELETE on +# exactly `auction` and `ah_worker_journal`. See "Worker DB grant delta" +# in doc/AuctionHouseBot.md. # Default: "" (disabled) # -# Example (point at a SELECT-only account, not mangosd's user): -# CharacterDatabaseInfo = "127.0.0.1;3306;ahbot_ro;change-me;character0" +# Example (point at the mode-appropriate dedicated account): +# CharacterDatabaseInfo = "127.0.0.1;3306;ahworker;change-me;character0" CharacterDatabaseInfo = "" @@ -103,4 +104,25 @@ AhBot.UpdateIntervalMs = 20000 AH.Service.TickMs = 1000 +# +# AH.Service.JournalPruneIntervalSec +# [SP-3] Worker-journal maintenance cadence in seconds. When +# WriteAuthority is active, the worker deletes terminal JRN_APPLIED rows +# older than AH.Service.JournalAppliedRetentionSec on this interval. +# Set to 0 to disable pruning. +# Default: 3600 + +AH.Service.JournalPruneIntervalSec = 3600 + +# +# AH.Service.JournalAppliedRetentionSec +# [SP-3] Retention window for terminal ah_worker_journal rows. Only +# JRN_APPLIED rows with resolved_time older than this many seconds are +# pruned; active COMMITTED, RESOLVING, CANCEL_PREPARED, and +# INTENT_PENDING rows are never deleted by this maintenance pass. Set to +# 0 to disable pruning. +# Default: 2592000 (30 days) + +AH.Service.JournalAppliedRetentionSec = 2592000 + ############################################################################### From f0e2e6fe8980e88b60c93b110f13b9fcbea26d64 Mon Sep 17 00:00:00 2001 From: MadMax Date: Wed, 15 Jul 2026 00:11:23 +0100 Subject: [PATCH 2/3] fix(ahworker): bound terminal journal pruning --- cmake/MangosParams.cmake | 2 +- doc/AuctionHouseBot.md | 4 +- src/modules/AhWorker/Journal.cpp | 110 +++++- src/modules/AhWorker/Journal.h | 23 +- src/modules/AhWorker/Main.cpp | 337 ++++++++++++++++--- src/modules/AhWorker/MutationHandler.cpp | 19 +- src/modules/AhWorker/ah-service.conf.dist.in | 28 +- src/shared/revision_data.h.in | 4 +- 8 files changed, 448 insertions(+), 79 deletions(-) diff --git a/cmake/MangosParams.cmake b/cmake/MangosParams.cmake index 92b697105..68383ba3c 100644 --- a/cmake/MangosParams.cmake +++ b/cmake/MangosParams.cmake @@ -2,4 +2,4 @@ set(MANGOS_EXP "CLASSIC") set(MANGOS_PKG "Mangos Zero") set(MANGOS_WORLD_VER 2026070700) set(MANGOS_REALM_VER 2026060300) -set(MANGOS_AHBOT_VER 2026070800) +set(MANGOS_AHBOT_VER 2026071400) diff --git a/doc/AuctionHouseBot.md b/doc/AuctionHouseBot.md index f8a957a9b..dd926b8d2 100644 --- a/doc/AuctionHouseBot.md +++ b/doc/AuctionHouseBot.md @@ -302,11 +302,13 @@ Enabling `WriteAuthority` with `Custody = 0` is unsupported. **Worker DB grant delta.** On top of the SELECT-only reader grants in *Security* above, a write-authority worker's DB account needs INSERT/UPDATE/DELETE on -exactly two Character-DB tables -- and nothing else: +exactly two Character-DB tables. Journal retention also reads resolution +idempotency markers from `custody_ledger`; it needs no custody write access: ```sql GRANT INSERT, UPDATE, DELETE ON `character0`.`auction` TO 'ahworker'@'localhost'; GRANT INSERT, UPDATE, DELETE ON `character0`.`ah_worker_journal` TO 'ahworker'@'localhost'; +GRANT SELECT ON `character0`.`custody_ledger` TO 'ahworker'@'localhost'; ``` **Residual mangosd-side writers that stand down (spec section 5.7).** With diff --git a/src/modules/AhWorker/Journal.cpp b/src/modules/AhWorker/Journal.cpp index 2fc3b9205..6ade59e55 100644 --- a/src/modules/AhWorker/Journal.cpp +++ b/src/modules/AhWorker/Journal.cpp @@ -24,6 +24,7 @@ #include "Database/DatabaseEnv.h" #include +#include namespace { @@ -71,6 +72,33 @@ namespace } return out; } + + bool CountPruneCandidates(ServiceDatabase& db, char const* ageColumn, + uint8 state, uint64 cutoff, uint32 limit, + bool protectResolveMarkers, uint32& count) + { + char const* markerClause = protectResolveMarkers + ? " AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`idem_key` = CONCAT('resolve:', `j`.`uuid`))" + : ""; + QueryResult* result = db.Character().PQuery( + "SELECT COUNT(*) FROM (" + "SELECT `j`.`uuid` FROM `ah_worker_journal` AS `j` " + "WHERE `j`.`state` = %u AND `j`.`%s` < %llu%s " + "ORDER BY `j`.`%s`, `j`.`uuid` LIMIT %u" + ") AS `prune_candidates`", + static_cast(state), ageColumn, + static_cast(cutoff), markerClause, + ageColumn, limit); + if (result == NULL) + { + return false; + } + + count = result->Fetch()[0].GetUInt32(); + delete result; + return true; + } } void AhJournal::Insert(ServiceDatabase& db, JournalRow const& row) @@ -98,6 +126,18 @@ void AhJournal::SetState(ServiceDatabase& db, uint64 uuid, uint8 state, static_cast(uuid)); } +void AhJournal::SetAppliedNow(ServiceDatabase& db, uint64 uuid) +{ + uint64 wallNow = static_cast(time(NULL)); + if (wallNow == std::numeric_limits::max()) + { + // Fail safe: retain the row instead of making a fresh APPLIED row look + // ancient to a wall-clock retention cutoff. + wallNow = UINT64_C(0x7FFFFFFFFFFFFFFF); + } + SetState(db, uuid, JRN_APPLIED, wallNow); +} + bool AhJournal::Get(ServiceDatabase& db, uint64 uuid, JournalRow& out) { QueryResult* result = db.Character().PQuery( @@ -181,10 +221,70 @@ void AhJournal::LoadActive(ServiceDatabase& db, std::vector& out) delete result; } -void AhJournal::DeleteAppliedOlderThan(ServiceDatabase& db, uint64 cutoff) +bool AhJournal::DeleteTerminalBatchOlderThan(ServiceDatabase& db, uint64 cutoff, + uint32 batchRows, bool& hasMore, + uint32& deletedRows) { - db.Character().DirectPExecute( - "DELETE FROM `ah_worker_journal` WHERE `state` = %u AND `resolved_time` < %llu", - static_cast(JRN_APPLIED), - static_cast(cutoff)); + hasMore = false; + deletedRows = 0u; + if (batchRows == 0u) + { + return true; + } + + uint32 committedCount = 0u; + uint32 appliedCount = 0u; + if (!CountPruneCandidates(db, "created_time", JRN_COMMITTED, cutoff, + batchRows, false, committedCount) || + !CountPruneCandidates(db, "resolved_time", JRN_APPLIED, cutoff, + batchRows, true, appliedCount)) + { + return false; + } + + if (committedCount == 0u && appliedCount == 0u) + { + return true; + } + + if (!db.Character().BeginTransaction()) + { + return false; + } + + bool queued = true; + if (committedCount != 0u) + { + queued = db.Character().PExecute( + "DELETE FROM `ah_worker_journal` " + "WHERE `state` = %u AND `created_time` < %llu " + "ORDER BY `created_time`, `uuid` LIMIT %u", + static_cast(JRN_COMMITTED), + static_cast(cutoff), committedCount); + } + if (queued && appliedCount != 0u) + { + queued = db.Character().PExecute( + "DELETE FROM `ah_worker_journal` " + "WHERE `state` = %u AND `resolved_time` < %llu " + "AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`idem_key` = CONCAT('resolve:', " + "`ah_worker_journal`.`uuid`)) " + "ORDER BY `resolved_time`, `uuid` LIMIT %u", + static_cast(JRN_APPLIED), + static_cast(cutoff), appliedCount); + } + if (!queued) + { + db.Character().RollbackTransaction(); + return false; + } + if (!db.Character().CommitTransactionChecked()) + { + return false; + } + + deletedRows = committedCount + appliedCount; + hasMore = committedCount == batchRows || appliedCount == batchRows; + return true; } diff --git a/src/modules/AhWorker/Journal.h b/src/modules/AhWorker/Journal.h index dbb01e050..1d2e2b761 100644 --- a/src/modules/AhWorker/Journal.h +++ b/src/modules/AhWorker/Journal.h @@ -69,6 +69,9 @@ namespace AhJournal /// Append a state/resolved_time UPDATE for @p uuid to the open txn. void SetState(ServiceDatabase& db, uint64 uuid, uint8 state, uint64 resolvedTime); + /// Append an APPLIED transition stamped from the worker wall clock. + void SetAppliedNow(ServiceDatabase& db, uint64 uuid); + /// Synchronous fetch by uuid. @return true if found. bool Get(ServiceDatabase& db, uint64 uuid, JournalRow& out); @@ -92,8 +95,24 @@ namespace AhJournal bool MaxSeqForRunId(ServiceDatabase& db, uint32 runId, bool highHalf, uint32& outMaxSeq); - /// Prune JRN_APPLIED rows with resolved_time < @p cutoff (standalone). - void DeleteAppliedOlderThan(ServiceDatabase& db, uint64 cutoff); + /** + * @brief Delete one bounded batch from each terminal journal state. + * + * JRN_COMMITTED age is measured by created_time. JRN_APPLIED age is + * measured by resolved_time, and rows with a matching resolve: + * custody marker are retained. At most @p batchRows rows from each state + * are deleted in one checked transaction. + * + * @param db Worker database facade. + * @param cutoff Delete terminal rows older than this Unix timestamp. + * @param batchRows Maximum rows deleted per terminal state. + * @param hasMore Set when either state filled its batch. + * @param deletedRows Set to the number selected for deletion. + * @return true on success, false on query or transaction failure. + */ + bool DeleteTerminalBatchOlderThan(ServiceDatabase& db, uint64 cutoff, + uint32 batchRows, bool& hasMore, + uint32& deletedRows); } #endif // AH_WORKER_JOURNAL_H diff --git a/src/modules/AhWorker/Main.cpp b/src/modules/AhWorker/Main.cpp index f038ac582..12c3cd739 100644 --- a/src/modules/AhWorker/Main.cpp +++ b/src/modules/AhWorker/Main.cpp @@ -88,9 +88,10 @@ #include #include -static const uint32 AH_JOURNAL_APPLIED_RETENTION_SEC_DEFAULT = +static const uint32 AH_JOURNAL_TERMINAL_RETENTION_SEC_DEFAULT = 30u * 24u * 60u * 60u; static const uint32 AH_JOURNAL_PRUNE_INTERVAL_SEC_DEFAULT = 60u * 60u; +static const uint32 AH_JOURNAL_PRUNE_BATCH_ROWS = 100u; static bool AhParseNonNegativeSeconds(std::string const& text, uint32& result) { @@ -120,11 +121,12 @@ static bool AhParseNonNegativeSeconds(std::string const& text, uint32& result) return true; } -static uint32 AhConfigNonNegativeSeconds(char const* name, uint32 defValue) +static uint32 AhConfigNonNegativeSeconds(Config& config, char const* name, + uint32 defValue) { char defText[16]; snprintf(defText, sizeof(defText), "%u", defValue); - std::string const text = sConfig.GetStringDefault(name, defText); + std::string const text = config.GetStringDefault(name, defText); uint32 value = 0u; if (AhParseNonNegativeSeconds(text, value)) @@ -137,6 +139,30 @@ static uint32 AhConfigNonNegativeSeconds(char const* name, uint32 defValue) return defValue; } +static uint32 AhConfigTerminalRetentionSeconds(Config& config, + uint32 defValue) +{ + char const* terminalName = "AH.Service.JournalTerminalRetentionSec"; + char const* legacyName = "AH.Service.JournalAppliedRetentionSec"; + char const* missing = "__MANGOS_AH_SETTING_NOT_FOUND__"; + std::string const terminalValue = + config.GetStringDefault(terminalName, missing); + if (terminalValue == missing) + { + return AhConfigNonNegativeSeconds(config, legacyName, defValue); + } + + uint32 value = 0u; + if (AhParseNonNegativeSeconds(terminalValue, value)) + { + return value; + } + + fprintf(stderr, "ah-service: invalid %s value '%s' - using default %u\n", + terminalName, terminalValue.c_str(), defValue); + return defValue; +} + static bool AhJournalPruneDue(uint64 now, uint64& nextPrune, uint32 intervalSec, uint32 retentionSec, uint64& cutoff) @@ -155,15 +181,15 @@ static bool AhJournalPruneDue(uint64 now, uint64& nextPrune, if (nextPrune == 0u) { nextPrune = now + intervalSec; - return false; } - - if (now < nextPrune) + else if (now < nextPrune) { return false; } - - nextPrune = now + intervalSec; + else + { + nextPrune = now + intervalSec; + } if (now <= retentionSec) { return false; @@ -1867,7 +1893,7 @@ static int RunWireSelfTest() /** * @brief Round-trip the AhJournal DAO against the configured character DB: * Insert (with a binary facts blob) -> Get -> SetState -> LoadActive - * membership -> DeleteAppliedOlderThan prune. Skips (returns 0) when + * membership -> bounded terminal prune. Skips (returns 0) when * no character DB is configured, mirroring the DB-less selftest rule. * @return 0 on success or skip, 1 on any assertion failure. */ @@ -1886,16 +1912,27 @@ static int RunJournalSelfTest() facts.push_back('\x00'); facts.push_back('\x01'); facts.push_back('\xFF'); facts.push_back('\x00'); facts.push_back('\x7A'); - uint64 const uuid = UINT64_C(0x00000009DEADBEEF); + uint64 const firstUuid = UINT64_C(0xFFFFFFFE00000010); + uint64 const lastUuid = UINT64_C(0xFFFFFFFE0000001F); + uint64 const uuid = firstUuid; + char resolveKey[64]; + snprintf(resolveKey, sizeof(resolveKey), "resolve:%llu", + static_cast(uuid)); + + // Clean any prior run before constructing the fixtures. + db.Character().DirectPExecute( + "DELETE FROM `ah_worker_journal` WHERE `uuid` BETWEEN %llu AND %llu", + static_cast(firstUuid), + static_cast(lastUuid)); + bool const canWriteCustodyFixture = db.Character().DirectPExecute( + "DELETE FROM `custody_ledger` WHERE `idem_key` = '%s'", resolveKey); + AhJournal::JournalRow row; row.uuid = uuid; row.auctionId = 12345u; row.kind = 0x41u; row.state = AhJournal::JRN_RESOLVING; row.facts = facts; - row.createdTime = 1000u; row.resolvedTime = 0u; + row.createdTime = 1u; row.resolvedTime = 0u; - // Clean any prior run, then Insert inside a checked txn. - db.Character().DirectPExecute( - "DELETE FROM `ah_worker_journal` WHERE `uuid` = %llu", - static_cast(uuid)); + // Insert inside a checked txn. db.Character().BeginTransaction(); AhJournal::Insert(db, row); if (!db.Character().CommitTransactionChecked()) @@ -1907,23 +1944,51 @@ static int RunJournalSelfTest() AhJournal::JournalRow got; if (!AhJournal::Get(db, uuid, got) || got.auctionId != 12345u || got.kind != 0x41u || got.state != AhJournal::JRN_RESOLVING || - got.facts != facts || got.createdTime != 1000u) + got.facts != facts || got.createdTime != 1u) { fprintf(stderr, "journal selftest FAILED: get/round-trip (facts %u bytes)\n", static_cast(got.facts.size())); return 1; } + // APPLIED retention timestamps use the same wall-clock domain as the + // prune cutoff, even before the first IPC_GAMETIME frame arrives. + AhJournal::JournalRow wallStamped = row; + wallStamped.uuid = firstUuid + 9u; + db.Character().BeginTransaction(); + AhJournal::Insert(db, wallStamped); + if (!db.Character().CommitTransactionChecked()) + { + fprintf(stderr, "journal selftest FAILED: wall-time fixture insert\n"); + return 1; + } + uint64 const wallBefore = static_cast(time(NULL)); + db.Character().BeginTransaction(); + AhJournal::SetAppliedNow(db, wallStamped.uuid); + if (!db.Character().CommitTransactionChecked()) + { + fprintf(stderr, "journal selftest FAILED: wall-time APPLIED commit\n"); + return 1; + } + uint64 const wallAfter = static_cast(time(NULL)); + if (!AhJournal::Get(db, wallStamped.uuid, got) || + got.state != AhJournal::JRN_APPLIED || + got.resolvedTime < wallBefore || got.resolvedTime > wallAfter) + { + fprintf(stderr, "journal selftest FAILED: APPLIED timestamp clock\n"); + return 1; + } + // SetState -> APPLIED with a resolved_time. db.Character().BeginTransaction(); - AhJournal::SetState(db, uuid, AhJournal::JRN_APPLIED, 2000u); + AhJournal::SetState(db, uuid, AhJournal::JRN_APPLIED, 2u); if (!db.Character().CommitTransactionChecked()) { fprintf(stderr, "journal selftest FAILED: setstate commit\n"); return 1; } if (!AhJournal::Get(db, uuid, got) || got.state != AhJournal::JRN_APPLIED || - got.resolvedTime != 2000u) + got.resolvedTime != 2u) { fprintf(stderr, "journal selftest FAILED: setstate read-back\n"); return 1; @@ -1943,13 +2008,157 @@ static int RunJournalSelfTest() } } - // DeleteAppliedOlderThan prunes it (resolved_time 2000 < cutoff 3000). - AhJournal::DeleteAppliedOlderThan(db, 3000u); - if (AhJournal::Get(db, uuid, got)) + // A live resolve: marker protects an old APPLIED row. This fixture + // requires more than the production worker's SELECT-only custody grant, so + // skip only this subcase when the configured selftest account cannot write + // custody_ledger. + if (canWriteCustodyFixture) + { + if (!db.Character().DirectPExecute( + "INSERT INTO `custody_ledger` " + "(`idem_key`,`kind`,`role`,`state`,`auction_id`," + "`created_time`,`resolved_time`) " + "VALUES ('%s', 0, 4, 1, 12345, 1, 2)", resolveKey)) + { + fprintf(stderr, "journal selftest FAILED: custody marker insert\n"); + return 1; + } + + bool hasMore = false; + uint32 deletedRows = 0u; + if (!AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 0u || !AhJournal::Get(db, uuid, got)) + { + fprintf(stderr, "journal selftest FAILED: custody marker did not protect APPLIED row\n"); + return 1; + } + + if (!db.Character().DirectPExecute( + "DELETE FROM `custody_ledger` WHERE `idem_key` = '%s'", + resolveKey) || + !AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 1u || AhJournal::Get(db, uuid, got)) + { + fprintf(stderr, "journal selftest FAILED: unprotected APPLIED row not pruned\n"); + return 1; + } + } + else + { + printf("journal custody-marker subtest SKIPPED (no custody write grant)\n"); + db.Character().DirectPExecute( + "DELETE FROM `ah_worker_journal` WHERE `uuid` = %llu", + static_cast(uuid)); + } + + // COMMITTED rows are terminal, but their resolved_time is historically 0; + // created_time is therefore their retention anchor. + std::vector rows; + AhJournal::JournalRow committed = row; + committed.state = AhJournal::JRN_COMMITTED; + committed.facts.clear(); + committed.resolvedTime = 0u; + + committed.uuid = firstUuid + 1u; + committed.createdTime = 1u; + rows.push_back(committed); + committed.uuid = firstUuid + 2u; + committed.createdTime = 20u; + rows.push_back(committed); + + db.Character().BeginTransaction(); + for (size_t i = 0; i < rows.size(); ++i) + { + AhJournal::Insert(db, rows[i]); + } + if (!db.Character().CommitTransactionChecked()) + { + fprintf(stderr, "journal selftest FAILED: committed fixture insert\n"); + return 1; + } + + bool hasMore = false; + uint32 deletedRows = 0u; + if (!AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 10u, hasMore, deletedRows) || hasMore || + deletedRows != 1u || + AhJournal::Get(db, firstUuid + 1u, got) || + !AhJournal::Get(db, firstUuid + 2u, got)) + { + fprintf(stderr, "journal selftest FAILED: COMMITTED age pruning\n"); + return 1; + } + + // A three-row backlog drains two rows, yields to the service loop, then + // finishes with one row on the next call. + rows.clear(); + for (uint32 i = 0u; i < 3u; ++i) + { + committed.uuid = firstUuid + 3u + i; + committed.createdTime = 1u + i; + rows.push_back(committed); + } + db.Character().BeginTransaction(); + for (size_t i = 0; i < rows.size(); ++i) + { + AhJournal::Insert(db, rows[i]); + } + if (!db.Character().CommitTransactionChecked() || + !AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || !hasMore || + deletedRows != 2u || + !AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 1u) + { + fprintf(stderr, "journal selftest FAILED: bounded backlog drain\n"); + return 1; + } + + // Non-terminal states remain untouched even with ancient timestamps. + rows.clear(); + uint8 const activeStates[] = { + AhJournal::JRN_RESOLVING, + AhJournal::JRN_CANCEL_PREPARED, + AhJournal::JRN_INTENT_PENDING + }; + for (uint32 i = 0u; i < 3u; ++i) { - fprintf(stderr, "journal selftest FAILED: prune left the row\n"); + AhJournal::JournalRow active = committed; + active.uuid = firstUuid + 6u + i; + active.state = activeStates[i]; + active.createdTime = 1u; + active.resolvedTime = 2u; + rows.push_back(active); + } + db.Character().BeginTransaction(); + for (size_t i = 0; i < rows.size(); ++i) + { + AhJournal::Insert(db, rows[i]); + } + if (!db.Character().CommitTransactionChecked() || + !AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 10u, hasMore, deletedRows) || hasMore || + deletedRows != 0u) + { + fprintf(stderr, "journal selftest FAILED: active-state prune guard\n"); return 1; } + for (uint32 i = 0u; i < 3u; ++i) + { + if (!AhJournal::Get(db, firstUuid + 6u + i, got)) + { + fprintf(stderr, "journal selftest FAILED: active row was pruned\n"); + return 1; + } + } + + db.Character().DirectPExecute( + "DELETE FROM `ah_worker_journal` WHERE `uuid` BETWEEN %llu AND %llu", + static_cast(firstUuid), + static_cast(lastUuid)); printf("journal selftest OK\n"); fflush(stdout); @@ -1958,6 +2167,8 @@ static int RunJournalSelfTest() static int RunJournalPruneSchedulerSelfTest() { + std::string const sourceBefore = sConfig.GetFilename(); + Config testConfig; char configPath[] = "ah-service-prune-selftest-XXXXXX"; ACE_HANDLE const configHandle = ACE_OS::mkstemp(configPath); if (configHandle == ACE_INVALID_HANDLE) @@ -1983,25 +2194,33 @@ static int RunJournalPruneSchedulerSelfTest() "AhTest.Zero = 0\n" "AhTest.Negative = -1\n" "AhTest.Malformed = 3600oops\n" - "AhTest.Overflow = 4294967296\n"); + "AhTest.Overflow = 4294967296\n" + "AH.Service.JournalAppliedRetentionSec = 0\n"); fclose(configFile); - if (!sConfig.SetSource(configPath)) + if (!testConfig.SetSource(configPath)) { remove(configPath); fprintf(stderr, "journal prune scheduler selftest FAILED:" " could not load config fixture\n"); return 1; } - remove(configPath); uint32 const fallback = 77u; - if (AhConfigNonNegativeSeconds("AhTest.Valid", fallback) != 3600u || - AhConfigNonNegativeSeconds("AhTest.Zero", fallback) != 0u || - AhConfigNonNegativeSeconds("AhTest.Missing", fallback) != fallback || - AhConfigNonNegativeSeconds("AhTest.Negative", fallback) != fallback || - AhConfigNonNegativeSeconds("AhTest.Malformed", fallback) != fallback || - AhConfigNonNegativeSeconds("AhTest.Overflow", fallback) != fallback) + if (AhConfigTerminalRetentionSeconds(testConfig, fallback) != 0u) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " legacy disable setting was not preserved\n"); + return 1; + } + remove(configPath); + + if (AhConfigNonNegativeSeconds(testConfig, "AhTest.Valid", fallback) != 3600u || + AhConfigNonNegativeSeconds(testConfig, "AhTest.Zero", fallback) != 0u || + AhConfigNonNegativeSeconds(testConfig, "AhTest.Missing", fallback) != fallback || + AhConfigNonNegativeSeconds(testConfig, "AhTest.Negative", fallback) != fallback || + AhConfigNonNegativeSeconds(testConfig, "AhTest.Malformed", fallback) != fallback || + AhConfigNonNegativeSeconds(testConfig, "AhTest.Overflow", fallback) != fallback) { fprintf(stderr, "journal prune scheduler selftest FAILED:" " unsafe config value did not use default\n"); @@ -2011,11 +2230,11 @@ static int RunJournalPruneSchedulerSelfTest() uint64 nextPrune = 0u; uint64 cutoff = 0u; - if (AhJournalPruneDue(1000u, nextPrune, 3600u, 100u, cutoff) || - nextPrune != 4600u) + if (!AhJournalPruneDue(1000u, nextPrune, 3600u, 100u, cutoff) || + cutoff != 900u || nextPrune != 4600u) { fprintf(stderr, "journal prune scheduler selftest FAILED:" - " first call should only schedule\n"); + " first call should schedule and prune\n"); return 1; } @@ -2074,6 +2293,13 @@ static int RunJournalPruneSchedulerSelfTest() return 1; } + if (sConfig.GetFilename() != sourceBefore) + { + fprintf(stderr, "journal prune scheduler selftest FAILED:" + " global config source changed\n"); + return 1; + } + printf("journal prune scheduler selftest OK\n"); return 0; } @@ -3501,6 +3727,12 @@ int main(int argc, char** argv) if (selfTest) { + if (cfgPath != NULL && !sConfig.SetSource(cfgPath)) + { + printf("ah-service selftest: config '%s' unavailable; DB-backed" + " tests may skip\n", cfgPath); + } + int rc = RunWireSelfTest(); if (rc != 0) { @@ -3837,12 +4069,14 @@ int main(int argc, char** argv) const uint32 mutTickMs = static_cast( sConfig.GetIntDefault("AH.Service.TickMs", 1000)); uint64 nextJournalPrune = 0u; + uint64 journalPruneCutoff = 0u; + bool journalPruneActive = false; const uint32 journalPruneIntervalSec = AhConfigNonNegativeSeconds( - "AH.Service.JournalPruneIntervalSec", + sConfig, "AH.Service.JournalPruneIntervalSec", AH_JOURNAL_PRUNE_INTERVAL_SEC_DEFAULT); - const uint32 journalAppliedRetentionSec = AhConfigNonNegativeSeconds( - "AH.Service.JournalAppliedRetentionSec", - AH_JOURNAL_APPLIED_RETENTION_SEC_DEFAULT); + const uint32 journalTerminalRetentionSec = AhConfigTerminalRetentionSeconds( + sConfig, + AH_JOURNAL_TERMINAL_RETENTION_SEC_DEFAULT); while (!stop) { @@ -4155,12 +4389,31 @@ int main(int argc, char** argv) // SP-2 Task 8: re-send / abandon in-flight bot-sell materializations. ahHandler->ResendStalePendingSells(wallNow); - uint64 journalCutoff = 0u; - if (AhJournalPruneDue(wallNow, nextJournalPrune, + if (!journalPruneActive && + AhJournalPruneDue(wallNow, nextJournalPrune, journalPruneIntervalSec, - journalAppliedRetentionSec, journalCutoff)) + journalTerminalRetentionSec, + journalPruneCutoff)) + { + journalPruneActive = true; + } + + if (journalPruneActive) { - AhJournal::DeleteAppliedOlderThan(botDb, journalCutoff); + bool hasMore = false; + uint32 deletedRows = 0u; + if (!AhJournal::DeleteTerminalBatchOlderThan( + botDb, journalPruneCutoff, + AH_JOURNAL_PRUNE_BATCH_ROWS, hasMore, deletedRows)) + { + fprintf(stderr, "ah-service: terminal journal prune batch" + " failed - retrying next interval\n"); + journalPruneActive = false; + } + else + { + journalPruneActive = hasMore; + } } } diff --git a/src/modules/AhWorker/MutationHandler.cpp b/src/modules/AhWorker/MutationHandler.cpp index a437932f1..f3b0d63be 100644 --- a/src/modules/AhWorker/MutationHandler.cpp +++ b/src/modules/AhWorker/MutationHandler.cpp @@ -674,8 +674,7 @@ PlayerMutationResult MutationHandler::OnCancelDecide(uint64 uuid, uint32 auction { return MakeResult(uuid, op, MUT_REJECTED, BOOK_ERR_DATABASE); } - AhJournal::SetState(*m_db, uuid, AhJournal::JRN_APPLIED, - static_cast(time(NULL))); + AhJournal::SetAppliedNow(*m_db, uuid); if (!m_db->Character().CommitTransactionChecked()) { // Journal still says CANCEL_PREPARED: keep the memory lock so @@ -903,9 +902,7 @@ bool MutationHandler::CommitTerminalApply(uint64 uuid, uint32 auctionId) return false; } db.PExecute("DELETE FROM `auction` WHERE `id` = '%u'", auctionId); - AhJournal::SetState(*m_db, uuid, - static_cast(AhJournal::JRN_APPLIED), - m_gameTimeNow); + AhJournal::SetAppliedNow(*m_db, uuid); return db.CommitTransactionChecked(); } @@ -920,9 +917,7 @@ bool MutationHandler::JournalMarkApplied(uint64 uuid) { return false; } - AhJournal::SetState(*m_db, uuid, - static_cast(AhJournal::JRN_APPLIED), - m_gameTimeNow); + AhJournal::SetAppliedNow(*m_db, uuid); return m_db->Character().CommitTransactionChecked(); } @@ -1674,9 +1669,7 @@ bool MutationHandler::PersistBotListing(BookRow const& row, uint64 uuid) } m_book.Insert(row); // appends the auction INSERT to this open txn + memory // Retire the PENDING intent row to JRN_APPLIED in the same txn. - AhJournal::SetState(*m_db, uuid, - static_cast(AhJournal::JRN_APPLIED), - static_cast(time(NULL))); + AhJournal::SetAppliedNow(*m_db, uuid); if (!db.CommitTransactionChecked()) { m_book.RollbackInsert(row.id); @@ -1696,8 +1689,6 @@ bool MutationHandler::RetireIntentPending(uint64 uuid) { return false; } - AhJournal::SetState(*m_db, uuid, - static_cast(AhJournal::JRN_APPLIED), - static_cast(time(NULL))); + AhJournal::SetAppliedNow(*m_db, uuid); return m_db->Character().CommitTransactionChecked(); } diff --git a/src/modules/AhWorker/ah-service.conf.dist.in b/src/modules/AhWorker/ah-service.conf.dist.in index 68f28024e..29d69665b 100644 --- a/src/modules/AhWorker/ah-service.conf.dist.in +++ b/src/modules/AhWorker/ah-service.conf.dist.in @@ -45,8 +45,9 @@ WorldDatabaseConnections = 1 # SECURITY: use a DEDICATED least-privilege MySQL account, not mangosd's # full read/write credentials. Read-only service mode needs SELECT only. # WriteAuthority additionally requires INSERT, UPDATE, and DELETE on -# exactly `auction` and `ah_worker_journal`. See "Worker DB grant delta" -# in doc/AuctionHouseBot.md. +# exactly `auction` and `ah_worker_journal`, plus SELECT on +# `custody_ledger` for safe journal retention. See "Worker DB grant +# delta" in doc/AuctionHouseBot.md. # Default: "" (disabled) # # Example (point at the mode-appropriate dedicated account): @@ -107,22 +108,25 @@ AH.Service.TickMs = 1000 # # AH.Service.JournalPruneIntervalSec # [SP-3] Worker-journal maintenance cadence in seconds. When -# WriteAuthority is active, the worker deletes terminal JRN_APPLIED rows -# older than AH.Service.JournalAppliedRetentionSec on this interval. -# Set to 0 to disable pruning. +# WriteAuthority is active, the worker starts a terminal-row retention +# pass on startup and then on this interval. Large backlogs drain in +# fixed batches between IPC dispatch passes. Set to 0 to disable pruning. # Default: 3600 AH.Service.JournalPruneIntervalSec = 3600 # -# AH.Service.JournalAppliedRetentionSec -# [SP-3] Retention window for terminal ah_worker_journal rows. Only -# JRN_APPLIED rows with resolved_time older than this many seconds are -# pruned; active COMMITTED, RESOLVING, CANCEL_PREPARED, and -# INTENT_PENDING rows are never deleted by this maintenance pass. Set to -# 0 to disable pruning. +# AH.Service.JournalTerminalRetentionSec +# [SP-3] Retention window for terminal ah_worker_journal rows. +# JRN_COMMITTED uses created_time; JRN_APPLIED uses resolved_time. +# APPLIED resolution rows remain protected while custody_ledger contains +# their resolve: idempotency marker. RESOLVING, CANCEL_PREPARED, +# and INTENT_PENDING rows are never deleted by this pass. Set to 0 to +# disable pruning. Configurations staged from an earlier version of this +# feature retain AH.Service.JournalAppliedRetentionSec as a fallback; +# this terminal setting takes precedence when both are present. # Default: 2592000 (30 days) -AH.Service.JournalAppliedRetentionSec = 2592000 +AH.Service.JournalTerminalRetentionSec = 2592000 ############################################################################### diff --git a/src/shared/revision_data.h.in b/src/shared/revision_data.h.in index 60a1e3e23..c610ec1f3 100644 --- a/src/shared/revision_data.h.in +++ b/src/shared/revision_data.h.in @@ -50,8 +50,8 @@ #define CHAR_DB_VERSION_NR "22" #define CHAR_DB_STRUCTURE_NR "5" - #define CHAR_DB_CONTENT_NR "2" - #define CHAR_DB_UPDATE_DESCRIPT "Add_Ah_Worker_Journal" + #define CHAR_DB_CONTENT_NR "3" + #define CHAR_DB_UPDATE_DESCRIPT "Add_Ah_Journal_Prune_Indexes" #define WORLD_DB_VERSION_NR "22" #define WORLD_DB_STRUCTURE_NR "5" From c9b236b9c14d9a0db0f399be956f1084f3cb08cd Mon Sep 17 00:00:00 2001 From: MadMax Date: Wed, 15 Jul 2026 00:49:24 +0100 Subject: [PATCH 3/3] fix(ahworker): retain journal reconciliation evidence --- src/modules/AhWorker/Journal.cpp | 42 +++++++-- src/modules/AhWorker/Journal.h | 9 +- src/modules/AhWorker/Main.cpp | 95 +++++++++++++++++++- src/modules/AhWorker/ah-service.conf.dist.in | 14 +-- 4 files changed, 141 insertions(+), 19 deletions(-) diff --git a/src/modules/AhWorker/Journal.cpp b/src/modules/AhWorker/Journal.cpp index 6ade59e55..f225830ef 100644 --- a/src/modules/AhWorker/Journal.cpp +++ b/src/modules/AhWorker/Journal.cpp @@ -28,6 +28,12 @@ namespace { + enum PruneProtection + { + PRUNE_PROTECT_COMMITTED_RECONCILE, + PRUNE_PROTECT_APPLIED_IDEMPOTENCY + }; + /// Encode binary @p in as lowercase ASCII hex (NUL-safe for SQL literals). std::string HexEncode(std::string const& in) { @@ -75,12 +81,24 @@ namespace bool CountPruneCandidates(ServiceDatabase& db, char const* ageColumn, uint8 state, uint64 cutoff, uint32 limit, - bool protectResolveMarkers, uint32& count) + PruneProtection protection, uint32& count) { - char const* markerClause = protectResolveMarkers - ? " AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " - "WHERE `c`.`idem_key` = CONCAT('resolve:', `j`.`uuid`))" - : ""; + char const* protectionClause = ""; + if (protection == PRUNE_PROTECT_COMMITTED_RECONCILE) + { + protectionClause = + " AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`auction_id` = `j`.`auction_id` " + "AND `c`.`state` = 0)"; + } + else if (protection == PRUNE_PROTECT_APPLIED_IDEMPOTENCY) + { + protectionClause = + " AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`idem_key` = CONCAT('resolve:', `j`.`uuid`)) " + "AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`idem_key` = CONCAT('botlist:', `j`.`uuid`))"; + } QueryResult* result = db.Character().PQuery( "SELECT COUNT(*) FROM (" "SELECT `j`.`uuid` FROM `ah_worker_journal` AS `j` " @@ -88,7 +106,7 @@ namespace "ORDER BY `j`.`%s`, `j`.`uuid` LIMIT %u" ") AS `prune_candidates`", static_cast(state), ageColumn, - static_cast(cutoff), markerClause, + static_cast(cutoff), protectionClause, ageColumn, limit); if (result == NULL) { @@ -235,9 +253,11 @@ bool AhJournal::DeleteTerminalBatchOlderThan(ServiceDatabase& db, uint64 cutoff, uint32 committedCount = 0u; uint32 appliedCount = 0u; if (!CountPruneCandidates(db, "created_time", JRN_COMMITTED, cutoff, - batchRows, false, committedCount) || + batchRows, PRUNE_PROTECT_COMMITTED_RECONCILE, + committedCount) || !CountPruneCandidates(db, "resolved_time", JRN_APPLIED, cutoff, - batchRows, true, appliedCount)) + batchRows, PRUNE_PROTECT_APPLIED_IDEMPOTENCY, + appliedCount)) { return false; } @@ -258,6 +278,9 @@ bool AhJournal::DeleteTerminalBatchOlderThan(ServiceDatabase& db, uint64 cutoff, queued = db.Character().PExecute( "DELETE FROM `ah_worker_journal` " "WHERE `state` = %u AND `created_time` < %llu " + "AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`auction_id` = `ah_worker_journal`.`auction_id` " + "AND `c`.`state` = 0) " "ORDER BY `created_time`, `uuid` LIMIT %u", static_cast(JRN_COMMITTED), static_cast(cutoff), committedCount); @@ -270,6 +293,9 @@ bool AhJournal::DeleteTerminalBatchOlderThan(ServiceDatabase& db, uint64 cutoff, "AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " "WHERE `c`.`idem_key` = CONCAT('resolve:', " "`ah_worker_journal`.`uuid`)) " + "AND NOT EXISTS (SELECT 1 FROM `custody_ledger` AS `c` " + "WHERE `c`.`idem_key` = CONCAT('botlist:', " + "`ah_worker_journal`.`uuid`)) " "ORDER BY `resolved_time`, `uuid` LIMIT %u", static_cast(JRN_APPLIED), static_cast(cutoff), appliedCount); diff --git a/src/modules/AhWorker/Journal.h b/src/modules/AhWorker/Journal.h index 1d2e2b761..f417be752 100644 --- a/src/modules/AhWorker/Journal.h +++ b/src/modules/AhWorker/Journal.h @@ -98,10 +98,11 @@ namespace AhJournal /** * @brief Delete one bounded batch from each terminal journal state. * - * JRN_COMMITTED age is measured by created_time. JRN_APPLIED age is - * measured by resolved_time, and rows with a matching resolve: - * custody marker are retained. At most @p batchRows rows from each state - * are deleted in one checked transaction. + * JRN_COMMITTED age is measured by created_time and rows for an auction + * with reserved custody are retained for lost-reply reconciliation. + * JRN_APPLIED age is measured by resolved_time; rows with a matching + * resolve: or botlist: custody marker are retained. At most + * @p batchRows rows from each state are deleted in one checked transaction. * * @param db Worker database facade. * @param cutoff Delete terminal rows older than this Unix timestamp. diff --git a/src/modules/AhWorker/Main.cpp b/src/modules/AhWorker/Main.cpp index 12c3cd739..5f3b350e2 100644 --- a/src/modules/AhWorker/Main.cpp +++ b/src/modules/AhWorker/Main.cpp @@ -1918,6 +1918,12 @@ static int RunJournalSelfTest() char resolveKey[64]; snprintf(resolveKey, sizeof(resolveKey), "resolve:%llu", static_cast(uuid)); + char botlistKey[64]; + snprintf(botlistKey, sizeof(botlistKey), "botlist:%llu", + static_cast(firstUuid + 10u)); + char committedKey[64]; + snprintf(committedKey, sizeof(committedKey), "test:jrn-reserve:%llu", + static_cast(firstUuid + 11u)); // Clean any prior run before constructing the fixtures. db.Character().DirectPExecute( @@ -1925,7 +1931,8 @@ static int RunJournalSelfTest() static_cast(firstUuid), static_cast(lastUuid)); bool const canWriteCustodyFixture = db.Character().DirectPExecute( - "DELETE FROM `custody_ledger` WHERE `idem_key` = '%s'", resolveKey); + "DELETE FROM `custody_ledger` WHERE `idem_key` IN ('%s', '%s', '%s')", + resolveKey, botlistKey, committedKey); AhJournal::JournalRow row; row.uuid = uuid; row.auctionId = 12345u; row.kind = 0x41u; @@ -2044,6 +2051,86 @@ static int RunJournalSelfTest() fprintf(stderr, "journal selftest FAILED: unprotected APPLIED row not pruned\n"); return 1; } + + AhJournal::JournalRow botlistApplied = row; + botlistApplied.uuid = firstUuid + 10u; + botlistApplied.auctionId = 12346u; + botlistApplied.state = AhJournal::JRN_APPLIED; + botlistApplied.facts.clear(); + botlistApplied.resolvedTime = 2u; + db.Character().BeginTransaction(); + AhJournal::Insert(db, botlistApplied); + if (!db.Character().CommitTransactionChecked() || + !db.Character().DirectPExecute( + "INSERT INTO `custody_ledger` " + "(`idem_key`,`kind`,`role`,`state`,`auction_id`," + "`created_time`,`resolved_time`) " + "VALUES ('%s', 1, 4, 0, 12346, 1, 0)", botlistKey)) + { + fprintf(stderr, "journal selftest FAILED: botlist fixture insert\n"); + return 1; + } + + if (!AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 0u || + !AhJournal::Get(db, botlistApplied.uuid, got)) + { + fprintf(stderr, "journal selftest FAILED: botlist marker did not protect APPLIED row\n"); + return 1; + } + + if (!db.Character().DirectPExecute( + "DELETE FROM `custody_ledger` WHERE `idem_key` = '%s'", + botlistKey) || + !AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 1u || + AhJournal::Get(db, botlistApplied.uuid, got)) + { + fprintf(stderr, "journal selftest FAILED: unprotected botlist APPLIED row not pruned\n"); + return 1; + } + + AhJournal::JournalRow reconcileCommitted = row; + reconcileCommitted.uuid = firstUuid + 11u; + reconcileCommitted.auctionId = 12347u; + reconcileCommitted.state = AhJournal::JRN_COMMITTED; + reconcileCommitted.facts.clear(); + reconcileCommitted.resolvedTime = 0u; + db.Character().BeginTransaction(); + AhJournal::Insert(db, reconcileCommitted); + if (!db.Character().CommitTransactionChecked() || + !db.Character().DirectPExecute( + "INSERT INTO `custody_ledger` " + "(`idem_key`,`kind`,`role`,`state`,`auction_id`," + "`created_time`,`resolved_time`) " + "VALUES ('%s', 0, 1, 0, 12347, 1, 0)", committedKey)) + { + fprintf(stderr, "journal selftest FAILED: reconcile fixture insert\n"); + return 1; + } + + if (!AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 0u || + !AhJournal::Get(db, reconcileCommitted.uuid, got)) + { + fprintf(stderr, "journal selftest FAILED: reserved custody did not protect COMMITTED row\n"); + return 1; + } + + if (!db.Character().DirectPExecute( + "UPDATE `custody_ledger` SET `state` = 1, `resolved_time` = 2 " + "WHERE `idem_key` = '%s'", committedKey) || + !AhJournal::DeleteTerminalBatchOlderThan( + db, 10u, 2u, hasMore, deletedRows) || hasMore || + deletedRows != 1u || + AhJournal::Get(db, reconcileCommitted.uuid, got)) + { + fprintf(stderr, "journal selftest FAILED: reconciled COMMITTED row not pruned\n"); + return 1; + } } else { @@ -2159,6 +2246,12 @@ static int RunJournalSelfTest() "DELETE FROM `ah_worker_journal` WHERE `uuid` BETWEEN %llu AND %llu", static_cast(firstUuid), static_cast(lastUuid)); + if (canWriteCustodyFixture) + { + db.Character().DirectPExecute( + "DELETE FROM `custody_ledger` WHERE `idem_key` IN ('%s', '%s', '%s')", + resolveKey, botlistKey, committedKey); + } printf("journal selftest OK\n"); fflush(stdout); diff --git a/src/modules/AhWorker/ah-service.conf.dist.in b/src/modules/AhWorker/ah-service.conf.dist.in index 29d69665b..5731ac8fe 100644 --- a/src/modules/AhWorker/ah-service.conf.dist.in +++ b/src/modules/AhWorker/ah-service.conf.dist.in @@ -119,12 +119,14 @@ AH.Service.JournalPruneIntervalSec = 3600 # AH.Service.JournalTerminalRetentionSec # [SP-3] Retention window for terminal ah_worker_journal rows. # JRN_COMMITTED uses created_time; JRN_APPLIED uses resolved_time. -# APPLIED resolution rows remain protected while custody_ledger contains -# their resolve: idempotency marker. RESOLVING, CANCEL_PREPARED, -# and INTENT_PENDING rows are never deleted by this pass. Set to 0 to -# disable pruning. Configurations staged from an earlier version of this -# feature retain AH.Service.JournalAppliedRetentionSec as a fallback; -# this terminal setting takes precedence when both are present. +# COMMITTED rows remain protected while their auction has RESERVED +# custody needed for lost-reply reconciliation. APPLIED rows remain +# protected while custody_ledger contains their resolve: or +# botlist: idempotency marker. RESOLVING, CANCEL_PREPARED, and +# INTENT_PENDING rows are never deleted by this pass. Set to 0 to disable +# pruning. Configurations staged from an earlier version of this feature +# retain AH.Service.JournalAppliedRetentionSec as a fallback; this +# terminal setting takes precedence when both are present. # Default: 2592000 (30 days) AH.Service.JournalTerminalRetentionSec = 2592000