perf(chat): sends re-read by origin, chat scans under the scan budget

A z_sendmany submit set off a burst of wallet scans before anything had
changed. Ops now re-read the wallet when they finish, by who started them:
the user's sends and dialog ops get a re-read plus one forced balance once
none of their ops is pending; client auto-shield re-reads unforced; chat
messages and splits refresh only the chat note view. A pending send no
longer forces scans, and chat txids no longer enter the tracked-send set.

Chat ingest skips stored messages before decrypting and caches
undecryptable ones per identity. The inbox scan runs when getwalletinfo's
txcount changed (a cheap probe, at most 2% duty) or a cost-scaled safety
net is due; the note-buffer scan runs on need. Both are admitted by the
scan budget, on a steady clock. The chat payer comes from a per-address
view of verified notes, with debits re-applied across in-flight scans;
splits are sized to one payer and only when they can refill the buffer.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-09 11:06:34 -05:00
co-authored by Claude Opus 5.5
parent f86cc4d6b7
commit b29a09eaa0
15 changed files with 1037 additions and 177 deletions
+282
View File
@@ -31,6 +31,7 @@
#include "services/wallet_scan_governor.h"
#include "services/scan_policy.h"
#include "services/connect_scan_gate.h"
#include "services/chat_scan_policy.h"
#include "services/network_refresh_service.h"
#include "services/refresh_scheduler.h"
#include "services/note_consolidation.h"
@@ -8776,10 +8777,15 @@ void testHushChatService()
svc.setIdentity(bob);
EXPECT_TRUE(svc.hasIdentity());
EXPECT_FALSE(svc.store().contains("tx0", 1));
EXPECT_EQ(svc.ingest(batch, {}, 1000), 2);
EXPECT_EQ(svc.decryptAttempts(), 2u);
EXPECT_EQ((int)svc.store().size(), 2);
EXPECT_EQ(svc.ingest(batch, {}, 1000), 0); // re-scan dedups
EXPECT_EQ(svc.decryptAttempts(), 2u); // ...without decrypting the stored ones again
EXPECT_EQ((int)svc.store().size(), 2);
EXPECT_TRUE(svc.store().contains("tx0", 1));
EXPECT_FALSE(svc.store().contains("tx0", 0));
std::vector<ChatMessage> conv = svc.store().conversation("conv-x");
EXPECT_EQ((int)conv.size(), 2);
@@ -8798,6 +8804,7 @@ void testHushChatService()
creq[0].payload_memo = "hi, add me";
creq[0].payload_position = 1;
EXPECT_EQ(svc.ingest(creq, {}), 1);
EXPECT_EQ(svc.decryptAttempts(), 2u); // a contact request isn't decrypted
std::vector<ChatMessage> cy = svc.store().conversation("conv-y");
EXPECT_EQ((int)cy.size(), 1);
EXPECT_TRUE(cy[0].kind == ChatMessageKind::ContactRequest);
@@ -8810,6 +8817,30 @@ void testHushChatService()
svc2.setIdentity(mallory);
EXPECT_EQ(svc2.ingest(batch, {}, 1000), 0);
EXPECT_TRUE(svc2.store().empty());
EXPECT_EQ(svc2.decryptAttempts(), 2u);
EXPECT_EQ(svc2.ingest(batch, {}, 1000), 0);
EXPECT_EQ(svc2.decryptAttempts(), 2u); // undecryptable: not tried again...
svc2.setIdentity(mallory);
EXPECT_EQ(svc2.ingest(batch, {}, 1000), 0);
EXPECT_EQ(svc2.decryptAttempts(), 4u); // ...until the identity is set again
// A junk payload is tried once; the same (txid, position) with another header gets its own try.
std::vector<HushChatTransactionMetadata> junk{batch[0]};
junk[0].txid = "txbad";
junk[0].payload_memo = "zz";
const auto before = svc.decryptAttempts();
EXPECT_EQ(svc.ingest(junk, {}, 1000), 0);
EXPECT_EQ(svc.ingest(junk, {}, 1000), 0);
EXPECT_EQ(svc.decryptAttempts(), before + 1);
junk[0].secretstream_header_hex = batch[1].secretstream_header_hex;
EXPECT_EQ(svc.ingest(junk, {}, 1000), 0);
EXPECT_EQ(svc.decryptAttempts(), before + 2);
// An erased conversation's messages aren't "stored" any more (a later scan re-ingests them).
svc.store().eraseConversation("conv-x");
EXPECT_FALSE(svc.store().contains("tx0", 1));
svc.store().clear();
EXPECT_FALSE(svc.store().contains("txc", 1));
}
// Phase 2: persistent, seed-encrypted store — round-trip, dedup, per-wallet isolation, and
@@ -8923,6 +8954,9 @@ void testHushChatDatabase()
EXPECT_EQ((int)conv.size(), 1);
EXPECT_EQ(conv[0].body, std::string("persisted!"));
EXPECT_EQ(conv[0].timestamp, (std::int64_t)4242);
svc.setIdentity(bob);
EXPECT_EQ(svc.ingest(batch, times), 0); // reloaded: no decrypt for what is already stored
EXPECT_EQ(svc.decryptAttempts(), 0u);
}
}
@@ -10886,6 +10920,12 @@ void testBalancePollDecisionAndVisibility()
EXPECT_TRUE(decideBalancePoll(x) == D::Skip); }
{ auto x = in; x.neverFetched = true; x.firstReadTried = true; x.background = true; // at the tip: as before
EXPECT_TRUE(decideBalancePoll(x) == D::Explicit); }
// A committed user send in the background waits for the window; the routine rules still apply.
{ auto x = in; x.forced = true; x.background = true; x.visible = false; x.sinceLastScanSec = 100.0;
EXPECT_TRUE(decideBalancePoll(x) == D::Skip); }
{ auto x = in; x.forced = true; x.background = true; x.visible = false; x.autoShieldArmed = true;
x.sinceLastScanSec = 130.0;
EXPECT_TRUE(decideBalancePoll(x) == D::AskGovernor); }
{ auto x = in; x.forced = true; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Explicit); }
{ auto x = in; x.neverFetched = true; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Explicit); }
{ auto x = in; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); }
@@ -11335,6 +11375,244 @@ void testAddressListOnlyCollector()
}
}
// C38: classify a poll's terminal ops by who started them.
void testOperationBatchPlan()
{
using Refresh = dragonx::services::NetworkRefreshService;
using Origin = Refresh::OperationOrigin;
using nlohmann::json;
const json reply = json::array({
json{{"id", "u1"}, {"status", "success"}, {"result", json{{"txid", "tx-u1"}}}},
json{{"id", "u2"}, {"status", "failed"}, {"error", json{{"message", "bad"}}}},
json{{"id", "u3"}, {"status", "success"}},
json{{"id", "a1"}, {"status", "success"}, {"result", json{{"txid", "tx-a1"}}}},
json{{"id", "c1"}, {"status", "success"}, {"result", json{{"txid", "tx-c1"}}}},
json{{"id", "c2"}, {"status", "cancelled"}},
});
const std::unordered_map<std::string, Origin> origins{{"a1", Origin::Auto}, {"c1", Origin::Chat},
{"c2", Origin::Chat}, {"c3", Origin::Chat}};
auto plan = [&](const std::vector<std::string>& ids) {
return Refresh::planOperationBatch(Refresh::parseOperationStatusPoll(reply, ids), origins);
};
auto all = plan({"u1", "u2", "u3", "a1", "c1", "c2", "c3"});
EXPECT_TRUE(all.userSucceeded && all.autoSucceeded && all.walletTerminal && all.chatTerminal);
auto a = plan({"a1"});
EXPECT_TRUE(a.autoSucceeded && a.walletTerminal && !a.userSucceeded && !a.chatTerminal);
for (const char* id : {"c1", "c2", "c3"}) { // a chat success, a cancelled one, one gone stale
auto c = plan({id});
EXPECT_TRUE(c.chatTerminal && !c.walletTerminal && !c.userSucceeded && !c.autoSucceeded);
}
auto failed = plan({"u2"});
EXPECT_TRUE(failed.walletTerminal && !failed.userSucceeded && !failed.autoSucceeded && !failed.chatTerminal);
EXPECT_TRUE(plan({"u3"}).userSucceeded); // a success without a txid counts
EXPECT_TRUE(plan({"u1"}).userSucceeded); // absent from the map: User
EXPECT_FALSE(Refresh::anyUserOperationPending({}, origins));
EXPECT_FALSE(Refresh::anyUserOperationPending({"c1"}, origins));
EXPECT_FALSE(Refresh::anyUserOperationPending({"a1", "c1"}, origins));
EXPECT_TRUE(Refresh::anyUserOperationPending({"x"}, origins));
EXPECT_TRUE(Refresh::anyUserOperationPending({"c1", "u9"}, origins));
}
// C79: when the chat scans are worth sending.
void testChatScanPolicy()
{
using namespace dragonx::services;
using nlohmann::json;
EXPECT_EQ(parseWalletTxCount(json::object()), (std::int64_t)-1);
EXPECT_EQ(parseWalletTxCount(json{{"txcount", "42"}}), (std::int64_t)-1);
EXPECT_EQ(parseWalletTxCount(json{{"txcount", -3}}), (std::int64_t)-1);
EXPECT_EQ(parseWalletTxCount(json{{"txcount", 4.5}}), (std::int64_t)-1);
EXPECT_EQ(parseWalletTxCount(json::array({1})), (std::int64_t)-1);
EXPECT_EQ(parseWalletTxCount(json{{"txcount", 42}}), (std::int64_t)42);
EXPECT_TRUE(chatProbeDue(0.0, 100.0, 100.0)); // no cost known yet
EXPECT_FALSE(chatProbeDue(50.0, 100.0, 101.0)); // 50 ms: at most every 2.5 s (2%)
EXPECT_TRUE(chatProbeDue(50.0, 100.0, 102.6));
EXPECT_FALSE(chatProbeDue(200.0, 100.0, 105.0)); // 200 ms: at most every 10 s
EXPECT_TRUE(chatProbeDue(200.0, 100.0, 110.0));
EXPECT_FALSE(chatProbeDue(30000.0, 100.0, 159.0)); // however slow: at least once a minute
EXPECT_TRUE(chatProbeDue(30000.0, 100.0, 160.0));
EXPECT_NEAR(chatInboxSafetyNetSec(0.0), 60.0, 1e-9);
EXPECT_NEAR(chatInboxSafetyNetSec(1000.0), 60.0, 1e-9);
EXPECT_NEAR(chatInboxSafetyNetSec(21000.0), 1050.0, 1e-9);
EXPECT_NEAR(chatNoteFallbackSec(21000.0), 420.0, 1e-9);
ChatInboxScanInputs in;
in.basisTxCount = 10;
in.txCount = 10;
in.nowSec = 129.0;
in.lastScanAtSec = 100.0;
EXPECT_FALSE(chatInboxScanNeeded(in));
{ auto x = in; x.basisTxCount = -1; EXPECT_TRUE(chatInboxScanNeeded(x)); }
{ auto x = in; x.txCount = -1; EXPECT_TRUE(chatInboxScanNeeded(x)); }
{ auto x = in; x.txCount = 11; EXPECT_TRUE(chatInboxScanNeeded(x)); }
{ auto x = in; x.nowSec = 160.0; EXPECT_TRUE(chatInboxScanNeeded(x)); }
// An hour of 2.5 s ticks: a quiet wallet scans about once a minute; every count change is scanned at once.
auto simulate = [](double net, int changes) {
std::int64_t count = 100, basis = -1;
double last = 0.0;
int scans = 0;
bool everyChangeScanned = true;
for (double t = 0.0; t < 3600.0; t += 2.5) {
bool changed = false;
for (int i = 1; i <= changes; ++i) {
if (t >= 300.0 * i && t < 300.0 * i + 2.5) { ++count; changed = true; }
}
ChatInboxScanInputs x;
x.basisTxCount = basis;
x.txCount = count;
x.nowSec = t;
x.lastScanAtSec = last;
x.safetyNetSec = net;
const bool need = chatInboxScanNeeded(x);
if (changed && !need) everyChangeScanned = false;
if (need) { ++scans; basis = count; last = t; }
}
return std::make_pair(scans, everyChangeScanned);
};
auto quiet = simulate(chatInboxSafetyNetSec(50.0), 0);
EXPECT_TRUE(quiet.first >= 59 && quiet.first <= 61);
auto busy = simulate(chatInboxSafetyNetSec(50.0), 10);
EXPECT_TRUE(busy.first <= 71 && busy.second);
EXPECT_TRUE(simulate(chatInboxSafetyNetSec(21000.0), 0).first <= 4);
ChatNoteScanInputs n;
n.modelSeen = true;
n.verifiedBudget = 10;
n.pipelineNotes = 10;
n.tipHeight = 100;
n.lastScanTip = 100;
n.nowSec = 100.0;
n.lastScanAtSec = 90.0;
n.tipChangedAtSec = 0.0;
{ auto x = n; x.rescanRequested = true; EXPECT_FALSE(shouldScanChatNotes(x)); } // nothing can use it
n.onChatPage = true;
EXPECT_FALSE(shouldScanChatNotes(n));
{ auto x = n; x.modelSeen = false; EXPECT_TRUE(shouldScanChatNotes(x)); }
{ auto x = n; x.modelSeen = false; x.lastScanAtSec = 98.0; EXPECT_FALSE(shouldScanChatNotes(x)); } // the floor
{ auto x = n; x.rescanRequested = true; EXPECT_TRUE(shouldScanChatNotes(x)); }
{ // A starved queue and a settled new tip.
auto x = n; x.onChatPage = false; x.queueWaiting = true; x.verifiedBudget = 0; x.tipHeight = 101;
x.tipChangedAtSec = 96.0;
EXPECT_TRUE(shouldScanChatNotes(x));
x.tipChangedAtSec = 98.0; // the notifier may not have caught up yet
EXPECT_FALSE(shouldScanChatNotes(x));
x.tipHeight = 100; // no new tip: only the fallback
x.tipChangedAtSec = 0.0;
EXPECT_FALSE(shouldScanChatNotes(x));
x.nowSec = 160.0;
EXPECT_TRUE(shouldScanChatNotes(x));
}
{ auto x = n; x.onChatPage = false; x.splitOutstanding = true; x.tipHeight = 101; x.tipChangedAtSec = 96.0;
EXPECT_TRUE(shouldScanChatNotes(x)); }
{ auto x = n; x.verifiedBudget = 7; x.tipHeight = 101; x.tipChangedAtSec = 96.0; EXPECT_TRUE(shouldScanChatNotes(x)); }
{ auto x = n; x.tipHeight = 101; x.tipChangedAtSec = 96.0; EXPECT_FALSE(shouldScanChatNotes(x)); } // 10/10
// An hour on the Chat page at 2 s ticks, a block every 36 s.
auto notesHour = [](int verified, int pipeline, double fallback) {
double last = 0.0;
std::int64_t scanTip = -1, tip = 0;
double tipAt = 0.0;
int scans = 0;
bool settled = true;
for (double t = 0.0; t < 3600.0; t += 2.0) {
const std::int64_t newTip = static_cast<std::int64_t>(t / 36.0);
if (newTip != tip) { tip = newTip; tipAt = t; }
ChatNoteScanInputs x;
x.onChatPage = true;
x.modelSeen = scans > 0;
x.verifiedBudget = verified;
x.pipelineNotes = pipeline;
x.tipHeight = tip;
x.lastScanTip = scanTip;
x.nowSec = t;
x.lastScanAtSec = last;
x.tipChangedAtSec = tipAt;
x.fallbackSec = fallback;
if (shouldScanChatNotes(x)) {
++scans;
last = t;
if (t - tipAt >= kChatTipSettleSec) scanTip = tip;
else if (scans > 1 && verified < pipeline) settled = false;
}
}
return std::make_pair(scans, settled);
};
auto unfunded = notesHour(0, 0, 60.0);
EXPECT_TRUE(unfunded.first >= 59 && unfunded.first <= 61);
auto healthy = notesHour(10, 10, 60.0);
EXPECT_TRUE(healthy.first >= 59 && healthy.first <= 61);
auto maturing = notesHour(7, 10, 60.0);
EXPECT_TRUE(maturing.first <= 102); // one per settled block, plus the first look (repeated once settled)
EXPECT_TRUE(notesHour(10, 10, chatNoteFallbackSec(20000.0)).first <= 10);
using dragonx::rpc::RpcFailureKind;
EXPECT_TRUE(chatScanOutcome(true, RpcFailureKind::None) == ScanOutcome::Ok);
EXPECT_TRUE(chatScanOutcome(false, RpcFailureKind::ReplyTimeout) == ScanOutcome::TimedOut);
EXPECT_TRUE(chatScanOutcome(false, RpcFailureKind::Deferred) == ScanOutcome::Skipped);
EXPECT_TRUE(chatScanOutcome(false, RpcFailureKind::QueueFull) == ScanOutcome::Skipped);
EXPECT_TRUE(chatScanOutcome(false, RpcFailureKind::AppError) == ScanOutcome::Failed);
EXPECT_TRUE(chatScanOutcome(false, RpcFailureKind::None) == ScanOutcome::Failed);
}
// S15: the chat payer from the verified-note view.
void testChooseChatPayer()
{
using namespace dragonx::services;
ChatPayerMap funds{{"zA", {5000, 1}}, {"zB", {20000, 2}}, {"zC", {20000, 1}}};
EXPECT_EQ(chooseChatPayer(funds, "zA", 10000), std::string("zB")); // tie with zC: the smaller address
EXPECT_EQ(chooseChatPayer(funds, "zC", 10000), std::string("zC"));
EXPECT_EQ(chooseChatPayer(funds, "zA", 30000), std::string());
EXPECT_EQ(chooseChatPayer(funds, "zX", 10000), std::string("zB"));
EXPECT_EQ(chooseChatPayer({}, "zA", 1), std::string());
EXPECT_EQ(chooseChatPayer(ChatPayerMap{{"zA", {20000, 0}}}, "", 1), std::string()); // no verified note
EXPECT_EQ(richestChatPayerZat(funds), (std::uint64_t)20000);
EXPECT_EQ(richestChatPayerZat(ChatPayerMap{{"zA", {90000, 0}}, {"zB", {10, 1}}}), (std::uint64_t)10);
ChatPayerMap d{{"zA", {30, 3}}, {"zB", {7, 1}}, {"zC", {50, 5}}};
debitChatPayer(d, "zA", ChatPayerSpend::OneNote);
EXPECT_TRUE(d["zA"].zat == 20 && d["zA"].notes == 2);
debitChatPayer(d, "zB", ChatPayerSpend::OneNote);
EXPECT_TRUE(d["zB"].zat == 0 && d["zB"].notes == 0);
debitChatPayer(d, "zC", ChatPayerSpend::All);
EXPECT_TRUE(d["zC"].zat == 0 && d["zC"].notes == 0);
const auto size = d.size();
debitChatPayer(d, "zQ", ChatPayerSpend::OneNote); // unknown: no-op
EXPECT_EQ(d.size(), size);
}
// The chat scans share the governor's one slot, and their streams never read the change gate.
void testChatScanGovernorStreams()
{
using namespace dragonx::services;
WalletScanGovernor gov;
ScanSignals a;
a.tipHash = "a";
a.mempoolKnown = true;
auto bal = gov.tryAdmit(ScanStream::Balance, true, a, 0.0);
EXPECT_TRUE(static_cast<bool>(bal));
EXPECT_FALSE(static_cast<bool>(gov.tryAdmit(ScanStream::ChatFast, true, a, 1.0)));
gov.complete(bal, 0.0, ScanOutcome::Ok, 1.0);
auto fast = gov.tryAdmit(ScanStream::ChatFast, true, a, 2.0);
EXPECT_TRUE(static_cast<bool>(fast));
gov.dropOrphans(ScanStream::ChatFast, 3.0); // its reply never came
auto notes = gov.tryAdmit(ScanStream::ChatNotes, true, a, 4.0);
EXPECT_TRUE(static_cast<bool>(notes));
gov.complete(notes, 10.0, ScanOutcome::Ok, 5.0);
ScanSignals b;
b.tipHash = "b";
b.mempoolKnown = true;
b.mempoolTxs = 9;
b.walletEpoch = 3;
EXPECT_FALSE(gov.needed(ScanStream::ChatNotes, b, 1e6)); // admission only
auto f2 = gov.beginExplicit(ScanStream::ChatFast, a, 6.0);
gov.complete(f2, 10.0, ScanOutcome::Ok, 7.0);
EXPECT_FALSE(gov.needed(ScanStream::ChatFast, b, 1e6));
auto old = gov.beginExplicit(ScanStream::ChatNotes, a, 8.0);
gov.reset();
gov.complete(old, 1e6, ScanOutcome::Ok, 9.0); // from before the reset: a no-op
EXPECT_TRUE(static_cast<bool>(gov.tryAdmit(ScanStream::Balance, true, a, 10.0)));
}
int main()
{
testConnectionConfig();
@@ -11358,6 +11636,7 @@ int main()
testParseMoneyZat();
testSpendableBalanceNeeded();
testOperationFilterParams();
testOperationBatchPlan();
testHistoryScanPolicy();
testShieldedHistoryPass();
testHistoryPassWithCollector();
@@ -11373,6 +11652,9 @@ int main()
testConnectionInfoSeeding();
testConnectScanGate();
testScanPolicy();
testChatScanPolicy();
testChooseChatPayer();
testChatScanGovernorStreams();
testAddressListOnlyCollector();
testNetworkRefreshResultModels();
testTrackedSendRetirement();