payout: one in-flight record per payment, and give it a real identity

A restart duplicated every in-flight payment, deterministically, and the
duplicates were load-bearing.

gQueuedPayouts is memory-only. ReconcilePayouts restored gPaying from disk but
left that set empty, and FlushMaturePayouts deduped only against it and
against the ledger's paidTxids -- never against gPaying itself. So a round
that had been exploded but not yet paid got a SECOND PayingRecord for every
unpaid recipient on the first flush after every restart. Another restart added
a third. The crash test saw exactly that: two byte-identical records for one
(block, address).

ForgetPayment and ReschedulePayment then found their record with SamePayment(),
a field-by-field comparison, which stops being an identity the moment two
records are equal. Every update landed in the first slot and the second was
never touched:

  - the duplicate's retryAfter stayed 0, so it ran in every 500 ms batch
    instead of once a minute;
  - each pass wrote retries = 0+1 into the FIRST slot, resetting its count, so
    the 1440-retry budget never completed and forfeit was unreachable;
  - the duplicate's fProbeWallet was never cleared, so FindRecordedPoolPayout
    -- a full mapWallet scan under cs_mapWallet -- ran twice a second per stuck
    recipient, forever, along with GetBalance() and, whenever funds allowed,
    SendMoney/CreateTransaction under cs_main.

That is a permanent hot loop holding cs_mapWallet and cs_main on a pool node,
starving RebuildJob() so job refresh and block submission stall. It is reached
by the ordinary state of a payment waiting for coin or for an unlocked wallet.

It did not double-pay only because a slot can reach SendMoney unprobed only
when fProbeWallet is already false, and any successful payment calls
ForgetPayment, which erases exactly the one slot that could be in that state.
That invariant is emergent, undocumented, and dies the moment anyone erases by
index or gives the record a stable identity -- which is precisely what this
commit does, so it is replaced by a structural one.

Both halves are fixed.

Not creating duplicates:
  - ReconcilePayouts rebuilds gQueuedPayouts from the restored gPaying, so the
    first flush after a restart no longer re-explodes rounds already in flight;
  - FlushMaturePayouts additionally refuses to append a record for a
    (round, address) that gPaying already holds. That check is the invariant
    proper -- the round-level set is an optimisation that can be stale, gPaying
    is the list being protected;
  - LoadPaying collapses duplicates already on disk as it reads them, merging
    the retry and deferral counts upwards, so a file written by the broken
    build does not inherit the pathology into the fixed one.

Real identity: PayingRecord gains an id, assigned under gPayoutExecMutex, and
ForgetPayment/ReschedulePayment match on it instead of on values. SamePayment
is deleted. The id is process-local and deliberately not persisted -- gPaying
is rebuilt from disk at every start, so it only has to name a slot within one
run, and reading ids back would mean trusting a file to contain no duplicates.

The exactly-once property is now the conjunction of four checks that each hold
on their own, rather than one accident:
  1. at most one gPaying record per (round, address), enforced at insert;
  2. the ledger's paidTxids, re-checked at the top of every attempt;
  3. the wallet probe, once per record, before its first send;
  4. state updates addressed by record id, so no slot can be skipped.

No on-disk format change: the id is not written, and duplicate records in an
existing file are read and collapsed.

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;