diff options
| -rw-r--r-- | src/rpc.cpp | 156 |
1 files changed, 79 insertions, 77 deletions
diff --git a/src/rpc.cpp b/src/rpc.cpp index e4939b3..f853e78 100644 --- a/src/rpc.cpp +++ b/src/rpc.cpp | |||
| @@ -1125,52 +1125,54 @@ static void ReconcilePayouts() | |||
| 1125 | { | 1125 | { |
| 1126 | VersionedBlob payingSnap; | 1126 | VersionedBlob payingSnap; |
| 1127 | { | 1127 | { |
| 1128 | std::lock_guard<std::mutex> exec(gPayoutExecMutex); | 1128 | std::lock_guard<std::mutex> exec(gPayoutExecMutex); |
| 1129 | std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); | 1129 | std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); |
| 1130 | 1130 | ||
| 1131 | size_t settled = 0; | 1131 | size_t settled = 0; |
| 1132 | int64 settledAmount = 0; | 1132 | int64 settledAmount = 0; |
| 1133 | for (auto it = gPaying.begin(); it != gPaying.end(); ) { | 1133 | for (auto it = gPaying.begin(); it != gPaying.end(); ) { |
| 1134 | bool paid = false; | 1134 | bool paid = false; |
| 1135 | for (const PendingPayout& pp : gPendingPayouts) | 1135 | for (const PendingPayout& pp : gPendingPayouts) |
| 1136 | if (SamePayoutRound(pp.blockHeight, pp.blockHash, | 1136 | if (SamePayoutRound(pp.blockHeight, pp.blockHash, |
| 1137 | it->blockHeight, it->blockHash) | 1137 | it->blockHeight, it->blockHash) |
| 1138 | && pp.paidTxids.count(it->address)) { | 1138 | && pp.paidTxids.count(it->address)) { |
| 1139 | paid = true; | 1139 | paid = true; |
| 1140 | break; | 1140 | break; |
| 1141 | } | ||
| 1142 | if (paid) { | ||
| 1143 | settledAmount += it->amount; | ||
| 1144 | settled++; | ||
| 1145 | it = gPaying.erase(it); | ||
| 1146 | } else { | ||
| 1147 | // Every restored record gets one wallet probe before it is | ||
| 1148 | // allowed to send: this is precisely the window where the | ||
| 1149 | // wallet may hold a transaction the ledger has not recorded. | ||
| 1150 | it->fProbeWallet = true; | ||
| 1151 | ++it; | ||
| 1141 | } | 1152 | } |
| 1142 | if (paid) { | ||
| 1143 | settledAmount += it->amount; | ||
| 1144 | settled++; | ||
| 1145 | it = gPaying.erase(it); | ||
| 1146 | } else { | ||
| 1147 | // Every restored record gets one wallet probe before it is allowed | ||
| 1148 | // to send: this is precisely the window where the wallet may hold | ||
| 1149 | // a transaction the ledger has not recorded yet. | ||
| 1150 | it->fProbeWallet = true; | ||
| 1151 | ++it; | ||
| 1152 | } | 1153 | } |
| 1153 | } | 1154 | if (settled) { |
| 1154 | if (settled) { | 1155 | LogPrint("payout", "[payout] %zu in-flight payment(s) totalling %s" |
| 1155 | LogPrint("payout", "[payout] %zu in-flight payment(s) totalling %s are" | 1156 | " are already recorded paid in the ledger -- not" |
| 1156 | " already recorded paid in the ledger -- not resending\n", | 1157 | " resending\n", |
| 1157 | settled, FormatMoney(settledAmount).c_str()); | 1158 | settled, FormatMoney(settledAmount).c_str()); |
| 1158 | } | 1159 | } |
| 1159 | 1160 | ||
| 1160 | // Rebuild the "already exploded" set from what survived. Without this the | 1161 | // Rebuild the "already exploded" set from what survived. Without this |
| 1161 | // first flush after a restart re-explodes every round still in flight. | 1162 | // the first flush after a restart re-explodes every round in flight. |
| 1162 | gQueuedPayouts.clear(); | 1163 | gQueuedPayouts.clear(); |
| 1163 | for (const PayingRecord& r : gPaying) | 1164 | for (const PayingRecord& r : gPaying) |
| 1164 | gQueuedPayouts.insert(PayoutIdForBlock(r.blockHeight, r.blockHash)); | 1165 | gQueuedPayouts.insert(PayoutIdForBlock(r.blockHeight, r.blockHash)); |
| 1165 | 1166 | ||
| 1166 | payingSnap = SerializePayingLocked(); | 1167 | payingSnap = SerializePayingLocked(); |
| 1167 | 1168 | ||
| 1168 | if (!gPaying.empty()) { | 1169 | if (!gPaying.empty()) { |
| 1169 | int64 owed = 0; | 1170 | int64 owed = 0; |
| 1170 | for (const auto& r : gPaying) owed += r.amount; | 1171 | for (const auto& r : gPaying) owed += r.amount; |
| 1171 | LogPrint("payout", "[payout] resuming %zu in-flight payment(s) totalling %s\n", | 1172 | LogPrint("payout", "[payout] resuming %zu in-flight payment(s)" |
| 1172 | gPaying.size(), FormatMoney(owed).c_str()); | 1173 | " totalling %s\n", |
| 1173 | } | 1174 | gPaying.size(), FormatMoney(owed).c_str()); |
| 1175 | } | ||
| 1174 | } | 1176 | } |
| 1175 | PublishPaying(payingSnap); | 1177 | PublishPaying(payingSnap); |
| 1176 | } | 1178 | } |
| @@ -1606,44 +1608,44 @@ static void FlushMaturePayouts() | |||
| 1606 | 1608 | ||
| 1607 | VersionedBlob payingSnap; | 1609 | VersionedBlob payingSnap; |
| 1608 | { | 1610 | { |
| 1609 | std::lock_guard<std::mutex> queue(gPayoutExecMutex); | 1611 | std::lock_guard<std::mutex> queue(gPayoutExecMutex); |
| 1610 | size_t added = 0; | 1612 | size_t added = 0; |
| 1611 | for (const PendingPayout& pp : ready) { | 1613 | for (const PendingPayout& pp : ready) { |
| 1612 | const std::string id = PayoutIdForBlock(pp.blockHeight, pp.blockHash); | 1614 | const std::string id = PayoutIdForBlock(pp.blockHeight, pp.blockHash); |
| 1613 | if (gQueuedPayouts.count(id)) | 1615 | if (gQueuedPayouts.count(id)) |
| 1614 | continue; // already exploded into gPaying and still in flight | 1616 | continue; // already exploded into gPaying and still in flight |
| 1615 | size_t n = 0; | 1617 | size_t n = 0; |
| 1616 | for (const auto& kv : pp.amounts) { | 1618 | for (const auto& kv : pp.amounts) { |
| 1617 | // Upstream's per-recipient dedupe: a recipient already carrying a | 1619 | // Upstream's per-recipient dedupe: a recipient already carrying |
| 1618 | // txid is done, even though the entry as a whole is not. | 1620 | // a txid is done, even though the entry as a whole is not. |
| 1619 | if (pp.paidTxids.count(kv.first)) | 1621 | if (pp.paidTxids.count(kv.first)) |
| 1620 | continue; | 1622 | continue; |
| 1621 | // The invariant, checked against the thing it is an invariant of: | 1623 | // The invariant, checked against the thing it is an invariant |
| 1622 | // at most one in-flight record per (round, address). The round-level | 1624 | // of: at most one in-flight record per (round, address). The |
| 1623 | // set above is an optimisation that can be wrong -- it is memory | 1625 | // round-level set above is an optimisation that can be wrong -- |
| 1624 | // only, it is rebuilt from gPaying at startup, and a stale `ready` | 1626 | // it is memory only, it is rebuilt from gPaying at startup, and |
| 1625 | // snapshot can race a round finishing -- whereas gPaying is the | 1627 | // a stale `ready` snapshot can race a round finishing -- |
| 1626 | // list being protected. A second record here is a second standing | 1628 | // whereas gPaying is the list being protected. A second record |
| 1627 | // order to send the same money. | 1629 | // here is a second standing order to send the same money. |
| 1628 | if (PaymentInFlightLocked(pp.blockHeight, pp.blockHash, kv.first)) | 1630 | if (PaymentInFlightLocked(pp.blockHeight, pp.blockHash, kv.first)) |
| 1629 | continue; | 1631 | continue; |
| 1630 | PayingRecord r; | 1632 | PayingRecord r; |
| 1631 | r.id = ++gPayingNextId; | 1633 | r.id = ++gPayingNextId; |
| 1632 | r.blockHeight = pp.blockHeight; | 1634 | r.blockHeight = pp.blockHeight; |
| 1633 | r.blockHash = pp.blockHash; | 1635 | r.blockHash = pp.blockHash; |
| 1634 | r.matureAtHeight = pp.matureAtHeight; | 1636 | r.matureAtHeight = pp.matureAtHeight; |
| 1635 | r.address = kv.first; | 1637 | r.address = kv.first; |
| 1636 | r.amount = kv.second; | 1638 | r.amount = kv.second; |
| 1637 | gPaying.push_back(r); | 1639 | gPaying.push_back(r); |
| 1638 | n++; | 1640 | n++; |
| 1641 | } | ||
| 1642 | gQueuedPayouts.insert(id); | ||
| 1643 | added += n; | ||
| 1644 | LogPrint("payout", "[payout] block height %d matured -- queuing" | ||
| 1645 | " payout %s (%zu of %zu recipients still owed)\n", | ||
| 1646 | pp.matureAtHeight, id.c_str(), n, pp.amounts.size()); | ||
| 1639 | } | 1647 | } |
| 1640 | gQueuedPayouts.insert(id); | 1648 | if (added) payingSnap = SerializePayingLocked(); |
| 1641 | added += n; | ||
| 1642 | LogPrint("payout", "[payout] block height %d matured -- queuing payout %s" | ||
| 1643 | " (%zu of %zu recipients still owed)\n", | ||
| 1644 | pp.matureAtHeight, id.c_str(), n, pp.amounts.size()); | ||
| 1645 | } | ||
| 1646 | if (added) payingSnap = SerializePayingLocked(); | ||
| 1647 | } | 1649 | } |
| 1648 | PublishPaying(payingSnap); | 1650 | PublishPaying(payingSnap); |
| 1649 | } | 1651 | } |