1 files changed, 97 insertions, 21 deletions

diff --git a/src/rpc.cpp b/src/rpc.cpp
index b6b26f1..7badb7c 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -680,6 +680,18 @@ static void LoadPendingPayouts()
 // and reconciling against the wallet on startup, is what makes a restart
 // survivable.
 struct PayingRecord {
+    // Identity of this SLOT, distinct from the identity of the payment it
+    // describes. ForgetPayment and ReschedulePayment used to find their record
+    // by comparing every field, which is only an identity while no two records
+    // can be equal -- and a restart used to produce byte-identical duplicates
+    // (see FlushMaturePayouts), so the first match absorbed every update and
+    // the second was never touched again.
+    //
+    // Process-local and deliberately not persisted: gPaying is rebuilt from
+    // disk at every start, so an id only ever has to name a slot within one
+    // run, and reading ids back would mean trusting a file to contain no
+    // duplicates. Assigned under gPayoutExecMutex from gPayingNextId.
+    uint64      id             = 0;
     int         blockHeight    = 0;
     uint256     blockHash      = 0; // zero = predates hash keying, see PendingPayout
     int         matureAtHeight = 0;
@@ -702,11 +714,37 @@ struct PayingRecord {
 };
 static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
 static std::mutex                gPayoutExecMutex;
+static uint64                    gPayingNextId = 0; // guarded by gPayoutExecMutex
 static uint64                    gPayingSeq = 0; // guarded by gPayoutExecMutex
 static std::mutex                gPayingFileMutex;
 static uint64                    gPayingPublishedSeq = 0; // guarded by gPayingFileMutex
 static std::atomic<bool>         gPayoutThreadRunning{false};
 
+// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
+// declared above with gPaying: the in-flight list IS the exec queue in this
+// tree, at per-recipient rather than per-round granularity, so a second queue
+// would be a second answer to "who is owed money". Only the dedupe set is new.
+//
+// Ids of rounds currently exploded into gPaying, so a flush cannot queue the
+// same round twice. Guarded by gPayoutExecMutex. Seeded from the restored
+// in-flight list by ReconcilePayouts -- it is memory-only, and leaving it
+// empty at startup was half of why a restart duplicated everything in flight.
+static std::set<std::string>      gQueuedPayouts;
+
+// Is this exact payment already in flight? gPaying is keyed on
+// (round, address): the ledger holds one amount per address per round, so two
+// records for that pair are two standing orders to send the same money.
+// Caller must hold gPayoutExecMutex.
+static bool PaymentInFlightLocked(int blockHeight, const uint256& blockHash,
+                                  const std::string& address)
+{
+    for (const PayingRecord& e : gPaying)
+        if (e.address == address
+            && SamePayoutRound(e.blockHeight, e.blockHash, blockHeight, blockHash))
+            return true;
+    return false;
+}
+
 static std::string PayingFilePath()
 {
     return GetAppDir() + "/paying_payouts.json";
@@ -774,6 +812,30 @@ static void LoadPaying()
                 r.retries = obj["retries"].get<int>();
             if (obj.contains("deferrals") && obj["deferrals"].is_number())
                 r.deferrals = obj["deferrals"].get<int>();
+            // Collapse duplicates as they are read. A build without the
+            // ReconcilePayouts seeding below wrote a second (and after another
+            // restart, a third) byte-identical record for every unpaid
+            // recipient, so a file written by one of those has to be cleaned
+            // up here or the pathology is inherited. The retry and deferral
+            // counts are merged upwards so collapsing does not hand the
+            // survivor a fresh budget.
+            PayingRecord* dup = NULL;
+            for (PayingRecord& e : gPaying)
+                if (e.address == r.address
+                    && SamePayoutRound(e.blockHeight, e.blockHash,
+                                       r.blockHeight, r.blockHash)) {
+                    dup = &e;
+                    break;
+                }
+            if (dup) {
+                dup->retries   = std::max(dup->retries, r.retries);
+                dup->deferrals = std::max(dup->deferrals, r.deferrals);
+                LogPrint("payout", "[payout] collapsing duplicate in-flight"
+                         " record for %s to %s -- one payment, one record\n",
+                         FormatMoney(r.amount).c_str(), r.address.c_str());
+                continue;
+            }
+            r.id = ++gPayingNextId;
             gPaying.push_back(r);
         }
         LogPrint("payout", "[payout] loaded %zu in-flight payment(s) from disk\n",
@@ -962,6 +1024,15 @@ static std::string PayoutComment(int blockHeight)
 // What is left is the useful part: the in-flight list restored from disk may
 // contain records the ledger already shows as paid (we died between the send
 // and the in-flight file being rewritten). The ledger wins.
+//
+// And the part that was missing: gQueuedPayouts is memory-only, so it came up
+// empty while gPaying came up full. FlushMaturePayouts deduped against that
+// empty set and against the ledger's paidTxids, and against nothing else --
+// so a round that was exploded but not yet paid got a SECOND record per
+// unpaid recipient on the first flush after every restart, deterministically.
+// (Observed live: two byte-identical records for one (block, address).) The
+// surviving in-flight list is exactly the statement of which rounds are
+// already exploded, so it is what the set is rebuilt from.
 static void ReconcilePayouts()
 {
     VersionedBlob payingSnap;
@@ -997,6 +1068,13 @@ static void ReconcilePayouts()
                  " already recorded paid in the ledger -- not resending\n",
                  settled, FormatMoney(settledAmount).c_str());
     }
+
+    // Rebuild the "already exploded" set from what survived. Without this the
+    // first flush after a restart re-explodes every round still in flight.
+    gQueuedPayouts.clear();
+    for (const PayingRecord& r : gPaying)
+        gQueuedPayouts.insert(PayoutIdForBlock(r.blockHeight, r.blockHash));
+
     payingSnap = SerializePayingLocked();
 
     if (!gPaying.empty()) {
@@ -1054,15 +1132,6 @@ static void MarkPoolRoundPaid(const std::string& id,
     PublishPoolRounds(snap);
 }
 
-// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
-// declared above with gPaying: the in-flight list IS the exec queue in this
-// tree, at per-recipient rather than per-round granularity, so a second queue
-// would be a second answer to "who is owed money". Only the dedupe set is new.
-//
-// Ids of rounds currently exploded into gPaying, so a flush cannot queue the
-// same round twice. Guarded by gPayoutExecMutex.
-static std::set<std::string>      gQueuedPayouts;
-
 static bool FindRecordedPoolPayout(const std::string& id,
                                    const std::string& address,
                                    int64 amount,
@@ -1311,7 +1380,17 @@ static void FlushMaturePayouts()
             // txid is done, even though the entry as a whole is not.
             if (pp.paidTxids.count(kv.first))
                 continue;
+            // The invariant, checked against the thing it is an invariant of:
+            // at most one in-flight record per (round, address). The round-level
+            // set above is an optimisation that can be wrong -- it is memory
+            // only, it is rebuilt from gPaying at startup, and a stale `ready`
+            // snapshot can race a round finishing -- whereas gPaying is the
+            // list being protected. A second record here is a second standing
+            // order to send the same money.
+            if (PaymentInFlightLocked(pp.blockHeight, pp.blockHash, kv.first))
+                continue;
             PayingRecord r;
+            r.id             = ++gPayingNextId;
             r.blockHeight    = pp.blockHeight;
             r.blockHash      = pp.blockHash;
             r.matureAtHeight = pp.matureAtHeight;
@@ -1331,27 +1410,24 @@ static void FlushMaturePayouts()
     PublishPaying(payingSnap);
 }
 
-// Identity of a payment, for finding its record again after SendMoney.
-// amounts is a map keyed by address, so (block, address) is already unique.
-static bool SamePayment(const PayingRecord& a, const PayingRecord& b)
-{
-    return SamePayoutRound(a.blockHeight, a.blockHash, b.blockHeight, b.blockHash)
-        && a.matureAtHeight == b.matureAtHeight
-        && a.address        == b.address
-        && a.amount         == b.amount;
-}
-
 // Settle a payment: it is either done or abandoned, either way it stops being
 // in flight. Note this only retires the working record -- whether the miner is
 // still OWED the money is the ledger's business, and the ledger is updated
 // separately by RecordPayoutTxid().
+//
+// Erases by record id, not by comparing fields. Value-matching was only an
+// identity for as long as no two records could be equal, and the restart bug
+// above produced records that were equal in every field: the first match then
+// absorbed every erase and every state update, and the second was never
+// touched again. The id names the slot the caller took its copy from, so the
+// update lands where it belongs whatever else is in the list.
 static void ForgetPayment(const PayingRecord& r)
 {
     VersionedBlob snap;
     {
         std::lock_guard<std::mutex> lk(gPayoutExecMutex);
         for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
-            if (SamePayment(*it, r)) { gPaying.erase(it); break; }
+            if (it->id == r.id) { gPaying.erase(it); break; }
 
         // When the last record for a round is gone the round is no longer in
         // flight, so a later flush may explode it again if the ledger still
@@ -1379,7 +1455,7 @@ static void ReschedulePayment(const PayingRecord& r)
     {
         std::lock_guard<std::mutex> lk(gPayoutExecMutex);
         for (auto& e : gPaying)
-            if (SamePayment(e, r)) {
+            if (e.id == r.id) {
                 e.retries      = r.retries;
                 e.deferrals    = r.deferrals;
                 e.retryAfter   = r.retryAfter;