perf(history): persist the sent-sweep watermark (windowed relaunch)

z_listsentbyaddress re-walks the whole wallet when there's no usable
watermark, which the daemon does under cs_main for every send -- hundreds
of seconds, 400+ s and ~19 GB RSS on a large wallet. The watermark lived
only in memory, so every relaunch, reconnect or daemon restart paid for a
full sweep. This persists it so those cases run a bounded windowed sweep
([z, 0, 2, tip - W + margin]) instead (SENT-HISTORY-SWEEP Part 1).

- The watermark (W = last swept tip), its anchor (A = W - margin) and anchor
  hash (H_A = getblockhash(A)), plus the enrichment backlog and the full-
  sweep backoff, ride in the encrypted history snapshot as an optional
  `sent_sweep` object (schema stays v1; a missing/inconsistent one reads as
  "no watermark" = today's behaviour, so older builds and payloads are safe).
- Before the first windowed sweep after a load, the anchor is verified
  (getblockhash(A) == H_A). A match means any fork is inside the window; a
  mismatch (reorg deeper than the margin, different chain/datadir) or an
  inability to check drops the watermark and sweeps the whole wallet.
- The first sweep after a load uses a bigger margin (absorbs notifier lag),
  and a watermark older than 2400 blocks uses the long timeout (callLong).
- The backlog is persisted WITH W: a pure z->z send's txid lives only in a
  sweep result and would be lost if the window moved past an un-enriched one.
- Watermark changes reach disk (folded into the store signature), and every
  hard case still forces a full sweep: reorg (anchor verify), GUI rescan /
  zap / import / address change, restoreOriginalWallet, and disconnect.

Deferred (follow-ups, each only turns a correct-but-slow case faster, never a
correctness risk): the banded full sweep for seeding on a pre-note-cache
daemon, the external-import tripwire, and the weekly safety-net re-sweep.
Part 2 (the daemon's decrypt cost) is already on the daemon's
perf/sapling-note-cache branch.

Pure helpers + cache round-trip + anchor-verify/large-window covered by
tests; the getblockhash-after-sweep updates the pinned RPC sequences.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/data/transaction_history_cache.cpp

@@ -366,6 +366,27 @@ TransactionHistoryCache::LoadResult TransactionHistoryCache::load(
                 }
             }
         }
+        // Sent-sweep watermark (optional). Drop it unless W/A are self-consistent and W is not above the
+        // stored tip (a replay after an unclean stop would otherwise use a stale watermark); the anchor
+        // hash is re-verified against the live chain on the next sweep regardless.
+        if (payload.contains("sent_sweep") && payload["sent_sweep"].is_object()) {
+            const auto& s = payload["sent_sweep"];
+            SentSweepState sw;
+            sw.height = s.value("height", -1);
+            sw.anchorHeight = s.value("anchor_height", -1);
+            sw.anchorHash = s.value("anchor_hash", std::string());
+            sw.fullFailures = s.value("full_failures", 0);
+            sw.fullRetryDelaySec = static_cast<long>(s.value("full_retry_delay_s", std::int64_t{0}));
+            sw.daemonId = s.value("daemon_id", std::string());
+            if (s.contains("backlog") && s["backlog"].is_array()) {
+                for (const auto& txid : s["backlog"])
+                    if (txid.is_string() && !txid.get<std::string>().empty())
+                        sw.backlog.push_back(txid.get<std::string>());
+            }
+            const bool consistent = sw.height >= 0 && sw.anchorHeight > 0 &&
+                                    sw.anchorHeight <= sw.height && sw.height <= tipHeight;
+            if (consistent) result.sentSweep = std::move(sw);
+        }
         result.tipHeight = tipHeight;
         result.tipHash = tipHash;
         result.updatedAt = updatedAt;
@@ -383,7 +404,8 @@ bool TransactionHistoryCache::replace(const std::string& walletIdentity,
                                       const std::string& tipHash,
                                       const std::vector<TransactionInfo>& transactions,
                                       std::time_t updatedAt,
-                                      const std::unordered_map<std::string, int>& shieldedScanHeights)
+                                      const std::unordered_map<std::string, int>& shieldedScanHeights,
+                                      const SentSweepState& sentSweep)
 {
     // The synchronous form (tests, tools): the same write the background writer does.
     SnapshotJob job;
@@ -394,6 +416,7 @@ bool TransactionHistoryCache::replace(const std::string& walletIdentity,
     job.transactions = transactions;
     job.updatedAt = updatedAt;
     job.shieldedScanHeights = shieldedScanHeights;
+    job.sentSweep = sentSweep;
     return writeSnapshot(std::move(job));
 }
 
@@ -426,6 +449,23 @@ bool TransactionHistoryCache::writeSnapshot(SnapshotJob&& job)
         }
     }
 
+    // Sent-sweep watermark, only when it is self-consistent (0 < A <= W <= tip). An unknown key means
+    // W = -1 on load, which is the pre-watermark behaviour, so older builds read a new payload safely.
+    const SentSweepState& sw = job.sentSweep;
+    if (sw.height >= 0 && sw.anchorHeight > 0 && sw.anchorHeight <= sw.height && sw.height <= job.tipHeight) {
+        json s;
+        s["height"] = sw.height;
+        s["anchor_height"] = sw.anchorHeight;
+        s["anchor_hash"] = sw.anchorHash;
+        s["backlog"] = json::array();
+        for (const auto& txid : sw.backlog)
+            if (!txid.empty()) s["backlog"].push_back(txid);
+        s["full_failures"] = sw.fullFailures;
+        s["full_retry_delay_s"] = static_cast<std::int64_t>(sw.fullRetryDelaySec);
+        s["daemon_id"] = sw.daemonId;
+        payload["sent_sweep"] = std::move(s);
+    }
+
     std::string plainText = payload.dump();
     std::vector<unsigned char> nonce;
     std::vector<unsigned char> cipherText;

src/data/transaction_history_cache.h

@@ -15,6 +15,24 @@ struct sqlite3;
 namespace dragonx {
 namespace data {
 
+// Persisted z_listsentbyaddress watermark, so a relaunch runs a windowed sweep instead of a full one.
+// height (W) = tip of the last successful sweep; anchorHeight (A) = W - margin, the bottom edge the window
+// covered; anchorHash (H_A) = getblockhash(A) at that time, verified on the next load. backlog = txids found
+// but not yet enriched (must persist WITH W, or a pure z->z send whose txid lives only in a sweep result is
+// lost once the window moves past it). The backoff fields keep a relaunch from immediately re-running a full
+// sweep that keeps failing; daemonId resets the backoff on a version change. (Namespace-scope, not nested, so
+// it is a complete type for the default argument of replace().)
+struct SentSweepState {
+    int height = -1;
+    int anchorHeight = -1;
+    std::string anchorHash;
+    std::vector<std::string> backlog;
+    int fullFailures = 0;
+    long fullRetryDelaySec = 0;
+    std::string daemonId;
+    bool valid() const { return height >= 0; }
+};
+
 class TransactionHistoryCache {
 public:
     struct LoadResult {
@@ -25,6 +43,7 @@ public:
         std::time_t updatedAt = 0;
         std::vector<TransactionInfo> transactions;
         std::unordered_map<std::string, int> shieldedScanHeights;
+        SentSweepState sentSweep;
     };
 
     // Everything one snapshot write needs, BY VALUE, so it can run on the history writer thread after the UI
@@ -38,6 +57,7 @@ public:
         std::vector<TransactionInfo> transactions;
         std::time_t updatedAt = 0;
         std::unordered_map<std::string, int> shieldedScanHeights;
+        SentSweepState sentSweep;
 
         SnapshotJob() = default;
         SnapshotJob(const SnapshotJob&) = delete;
@@ -81,7 +101,8 @@ public:
                  const std::string& tipHash,
                  const std::vector<TransactionInfo>& transactions,
                  std::time_t updatedAt,
-                 const std::unordered_map<std::string, int>& shieldedScanHeights = {});
+                 const std::unordered_map<std::string, int>& shieldedScanHeights = {},
+                 const SentSweepState& sentSweep = SentSweepState{});
     // Thread-safe snapshot write (the history writer thread; replace() wraps it): JSON + AEAD with no lock
     // held, then INSERT OR REPLACE + prune under the DB lock. Always wipes job.key. False for a keyless job.
     bool writeSnapshot(SnapshotJob&& job);

src/data/transaction_history_store_policy.cpp

@@ -65,7 +65,11 @@ HistorySnapshotSig historySnapshotSignature(const std::vector<TransactionInfo>&
                                             std::uint64_t keyId,
                                             const std::unordered_map<std::string, int>& shieldedScanHeights,
                                             int tipHeight,
-                                            const std::string& tipHash)
+                                            const std::string& tipHash,
+                                            int sentHeight,
+                                            int sentAnchorHeight,
+                                            const std::string& sentAnchorHash,
+                                            const std::unordered_set<std::string>& sentBacklog)
 {
     HistorySnapshotSig sig;
     std::uint64_t rowSum = 0;     // sums: commutative, so row order does not matter; duplicates still count
@@ -89,11 +93,19 @@ HistorySnapshotSig historySnapshotSignature(const std::vector<TransactionInfo>&
         exactSum += finalize(e);
         ++sig.rows;
     }
+    std::uint64_t backlogSum = 0;   // order-independent: a set, so sum is commutative
+    for (const auto& txid : sentBacklog) {
+        if (txid.empty()) continue;
+        std::uint64_t h = kOffset;
+        mixStr(h, txid);
+        backlogSum += finalize(h);
+    }
     std::uint64_t c = kOffset;
     mixU64(c, rowSum);
     mixU64(c, sig.rows);
     mixStr(c, walletHash);
     mixU64(c, keyId);
+    mixU64(c, backlogSum);
     sig.content = finalize(c);
 
     std::uint64_t heightSum = 0;
@@ -109,6 +121,9 @@ HistorySnapshotSig historySnapshotSignature(const std::vector<TransactionInfo>&
     mixU64(s, exactSum);
     mixU64(s, static_cast<std::uint64_t>(static_cast<std::int64_t>(tipHeight)));
     mixStr(s, tipHash);
+    mixU64(s, static_cast<std::uint64_t>(static_cast<std::int64_t>(sentHeight)));
+    mixU64(s, static_cast<std::uint64_t>(static_cast<std::int64_t>(sentAnchorHeight)));
+    mixStr(s, sentAnchorHash);
     sig.slow = finalize(s);
     return sig;
 }

src/data/transaction_history_store_policy.h

@@ -38,7 +38,15 @@ HistorySnapshotSig historySnapshotSignature(const std::vector<TransactionInfo>&
                                             std::uint64_t keyId,
                                             const std::unordered_map<std::string, int>& shieldedScanHeights,
                                             int tipHeight,
-                                            const std::string& tipHash);
+                                            const std::string& tipHash,
+                                            // Sent-sweep watermark: W/A/H_A move per sweep (folded into `slow`,
+                                            // written on the slow cadence and always on disconnect/lock/exit),
+                                            // the enrichment backlog into `content` (written within ~20 s), so an
+                                            // invalidation that only changes the watermark still reaches disk.
+                                            int sentHeight = -1,
+                                            int sentAnchorHeight = -1,
+                                            const std::string& sentAnchorHash = std::string(),
+                                            const std::unordered_set<std::string>& sentBacklog = {});
 
 enum class HistoryStoreMode {
     Coalesce,   // routine refresh: content changes at most every kHistoryMinStoreIntervalSec, slow drift every