diff --git a/res/lang/de.json b/res/lang/de.json index 0dc8224..b4adb69 100644 --- a/res/lang/de.json +++ b/res/lang/de.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "Max. Inputs pro Batch", "merge_no_spendable": "Noch keine verfügbaren Inputs zum Konsolidieren.", "merge_progress": "Konsolidierung läuft… dies kann einige Minuten dauern. Sie können dieses Fenster schließen.", + "merge_run_aborted_toast": "Die Verbindung zum Node wurde zurückgesetzt, daher wurde die Konsolidierung gestoppt. Führen Sie sie erneut aus, um fortzufahren.", + "merge_run_busy": "Eine Konsolidierung läuft. Warten Sie, bis sie fertig ist, oder stoppen Sie sie zuerst.", + "merge_run_chip_fmt": "Konsolidierung %d/%d", + "merge_run_err_timeout": "Der Node hat nicht rechtzeitig geantwortet; der Durchgang ist möglicherweise trotzdem eingereiht.", + "merge_run_halted_fmt": "Konsolidierung nach wiederholten Fehlern gestoppt: %s", + "merge_run_migration_pending": "Schließen Sie die Seed-Migration ab oder brechen Sie sie ab, bevor Sie konsolidieren.", + "merge_run_progress_fmt": "Konsolidierung läuft… Durchgang %d von etwa %d, bisher %d Notes zusammengeführt. Sie können dieses Fenster schließen; der Vorgang läuft im Hintergrund weiter.", + "merge_run_start_failed": "Die Konsolidierung konnte nicht gestartet werden – der Node ist nicht verbunden.", + "merge_run_status_not_run": "%s – nicht konsolidiert (%d Notes)", + "merge_run_status_waiting": "%s – wartet (%d Notes)", + "merge_run_stop": "Stoppen", + "merge_run_stopped_toast": "Konsolidierung gestoppt. Führen Sie sie erneut aus, um den Rest zusammenzuführen.", + "merge_run_stopping": "Wird nach den bereits gesendeten Durchgängen gestoppt…", + "merge_run_wait_batch": "Die Durchgänge werden verteilt, damit der Node reaktionsfähig bleibt…", + "merge_run_wait_send": "Pausiert, bis Ihre ausstehende Transaktion durch ist.", + "merge_run_wait_sync": "Pausiert: Der Node synchronisiert noch.", "merge_scope_failed_fmt": "Ihre Inputs konnten nicht gelesen werden: %s", "merge_scope_fmt": "%d transparente + %d geschützte Inputs · ~%s DRGX verfügbar", "merge_scope_loading": "Ihre Inputs werden geprüft…", diff --git a/res/lang/es.json b/res/lang/es.json index 4984d69..fca567a 100644 --- a/res/lang/es.json +++ b/res/lang/es.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "Máximo de entradas por lote", "merge_no_spendable": "Aún no hay entradas disponibles para consolidar.", "merge_progress": "Consolidando… esto puede tardar varios minutos. Puedes cerrar esta ventana.", + "merge_run_aborted_toast": "La conexión con el nodo se reinició, así que la consolidación se detuvo. Ejecútala de nuevo para continuar.", + "merge_run_busy": "Hay una consolidación en curso. Espera a que termine o detenla primero.", + "merge_run_chip_fmt": "Consolidando %d/%d", + "merge_run_err_timeout": "El nodo no respondió a tiempo; puede que el lote siga en cola.", + "merge_run_halted_fmt": "Consolidación detenida tras fallos repetidos: %s", + "merge_run_migration_pending": "Termina o cancela la migración de la semilla antes de consolidar.", + "merge_run_progress_fmt": "Consolidando… lote %d de unos %d, %d notas fusionadas hasta ahora. Puedes cerrar esta ventana; seguirá en segundo plano.", + "merge_run_start_failed": "No se pudo iniciar la consolidación: el nodo no está conectado.", + "merge_run_status_not_run": "%s: sin consolidar (%d notas)", + "merge_run_status_waiting": "%s: en espera (%d notas)", + "merge_run_stop": "Detener", + "merge_run_stopped_toast": "Consolidación detenida. Ejecútala de nuevo para fusionar el resto.", + "merge_run_stopping": "Deteniendo tras los lotes ya enviados…", + "merge_run_wait_batch": "Espaciando los lotes para que el nodo siga respondiendo…", + "merge_run_wait_send": "En pausa hasta que se complete tu transacción pendiente.", + "merge_run_wait_sync": "En pausa: el nodo aún se está sincronizando.", "merge_scope_failed_fmt": "No se pudieron leer tus entradas: %s", "merge_scope_fmt": "%d entradas transparentes + %d blindadas · ~%s DRGX disponibles", "merge_scope_loading": "Comprobando tus entradas…", diff --git a/res/lang/fr.json b/res/lang/fr.json index d50158e..2f6ee6b 100644 --- a/res/lang/fr.json +++ b/res/lang/fr.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "Nombre max d'entrées par lot", "merge_no_spendable": "Aucune entrée dépensable à consolider pour l'instant.", "merge_progress": "Consolidation en cours… cela peut prendre quelques minutes. Vous pouvez fermer cette fenêtre.", + "merge_run_aborted_toast": "La connexion au nœud a été réinitialisée, la consolidation s'est donc arrêtée. Relancez-la pour continuer.", + "merge_run_busy": "Une consolidation est en cours. Attendez qu'elle se termine ou arrêtez-la d'abord.", + "merge_run_chip_fmt": "Consolidation %d/%d", + "merge_run_err_timeout": "Le nœud n'a pas répondu à temps ; le lot est peut-être encore en file d'attente.", + "merge_run_halted_fmt": "Consolidation arrêtée après des échecs répétés : %s", + "merge_run_migration_pending": "Terminez ou annulez la migration de la phrase de récupération avant de consolider.", + "merge_run_progress_fmt": "Consolidation en cours… lot %d sur environ %d, %d notes fusionnées jusqu'ici. Vous pouvez fermer cette fenêtre ; l'opération continue en arrière-plan.", + "merge_run_start_failed": "Impossible de démarrer la consolidation — le nœud n'est pas connecté.", + "merge_run_status_not_run": "%s — non consolidé (%d notes)", + "merge_run_status_waiting": "%s — en attente (%d notes)", + "merge_run_stop": "Arrêter", + "merge_run_stopped_toast": "Consolidation arrêtée. Relancez-la pour fusionner le reste.", + "merge_run_stopping": "Arrêt après les lots déjà envoyés…", + "merge_run_wait_batch": "Espacement des lots pour que le nœud reste réactif…", + "merge_run_wait_send": "En pause jusqu'à ce que votre transaction en attente soit passée.", + "merge_run_wait_sync": "En pause : le nœud est encore en cours de synchronisation.", "merge_scope_failed_fmt": "Impossible de lire vos entrées : %s", "merge_scope_fmt": "%d entrées transparentes + %d blindées · ~%s DRGX dépensables", "merge_scope_loading": "Vérification de vos entrées…", diff --git a/res/lang/ja.json b/res/lang/ja.json index c9fe295..562106c 100644 --- a/res/lang/ja.json +++ b/res/lang/ja.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "バッチあたりの最大入力数", "merge_no_spendable": "統合できる使用可能な入力がまだありません。", "merge_progress": "統合中… 数分かかることがあります。このウィンドウは閉じても構いません。", + "merge_run_aborted_toast": "ノードとの接続がリセットされたため、統合を停止しました。続けるには再度実行してください。", + "merge_run_busy": "統合を実行中です。完了を待つか、先に停止してください。", + "merge_run_chip_fmt": "統合中 %d/%d", + "merge_run_err_timeout": "ノードが時間内に応答しませんでした。この回はまだキューに入っている可能性があります。", + "merge_run_halted_fmt": "失敗が続いたため統合を停止しました：%s", + "merge_run_migration_pending": "統合する前に、シードの移行を完了するかキャンセルしてください。", + "merge_run_progress_fmt": "統合中… %d 回目（全体で約 %d 回）、これまでに %d 件のノートを統合しました。このウィンドウを閉じてもバックグラウンドで続行します。", + "merge_run_start_failed": "統合を開始できませんでした — ノードが接続されていません。", + "merge_run_status_not_run": "%s — 未統合（%d 件のノート）", + "merge_run_status_waiting": "%s — 待機中（%d 件のノート）", + "merge_run_stop": "停止", + "merge_run_stopped_toast": "統合を停止しました。残りを統合するには再度実行してください。", + "merge_run_stopping": "送信済みの回が終わったら停止します…", + "merge_run_wait_batch": "ノードの応答性を保つため、間隔を空けて実行しています…", + "merge_run_wait_send": "保留中の取引が完了するまで一時停止しています。", + "merge_run_wait_sync": "一時停止中：ノードはまだ同期中です。", "merge_scope_failed_fmt": "入力を読み取れませんでした: %s", "merge_scope_fmt": "透明%d件 + シールド%d件の入力　·　使用可能 約%s DRGX", "merge_scope_loading": "入力を確認中…", diff --git a/res/lang/ko.json b/res/lang/ko.json index 6b6b996..5bcc361 100644 --- a/res/lang/ko.json +++ b/res/lang/ko.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "배치당 최대 입력 수", "merge_no_spendable": "아직 통합할 사용 가능한 입력이 없습니다.", "merge_progress": "통합 중… 몇 분 정도 걸릴 수 있습니다. 이 창을 닫아도 됩니다.", + "merge_run_aborted_toast": "노드 연결이 재설정되어 통합이 중지되었습니다. 계속하려면 다시 실행하세요.", + "merge_run_busy": "통합이 진행 중입니다. 완료될 때까지 기다리거나 먼저 중지하세요.", + "merge_run_chip_fmt": "통합 중 %d/%d", + "merge_run_err_timeout": "노드가 제때 응답하지 않았습니다. 배치가 아직 대기열에 있을 수 있습니다.", + "merge_run_halted_fmt": "반복된 실패로 통합이 중지되었습니다: %s", + "merge_run_migration_pending": "통합하기 전에 시드 마이그레이션을 완료하거나 취소하세요.", + "merge_run_progress_fmt": "통합 중… 배치 %d(약 %d개 중), 지금까지 노트 %d개 병합됨. 이 창을 닫아도 백그라운드에서 계속 진행됩니다.", + "merge_run_start_failed": "통합을 시작할 수 없습니다 — 노드가 연결되어 있지 않습니다.", + "merge_run_status_not_run": "%s — 통합 안 됨(노트 %d개)", + "merge_run_status_waiting": "%s — 대기 중(노트 %d개)", + "merge_run_stop": "중지", + "merge_run_stopped_toast": "통합이 중지되었습니다. 나머지를 병합하려면 다시 실행하세요.", + "merge_run_stopping": "이미 보낸 배치가 끝나면 중지합니다…", + "merge_run_wait_batch": "노드가 응답성을 유지하도록 배치 간격을 두는 중…", + "merge_run_wait_send": "보류 중인 거래가 완료될 때까지 일시 중지됩니다.", + "merge_run_wait_sync": "일시 중지됨: 노드가 아직 동기화 중입니다.", "merge_scope_failed_fmt": "입력을 읽을 수 없습니다: %s", "merge_scope_fmt": "투명 입력 %d개 + 보호 입력 %d개 · 사용 가능 약 %s DRGX", "merge_scope_loading": "입력을 확인하는 중…", diff --git a/res/lang/pt.json b/res/lang/pt.json index 1e10cad..4a98da7 100644 --- a/res/lang/pt.json +++ b/res/lang/pt.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "Máx. de entradas por lote", "merge_no_spendable": "Ainda não há entradas disponíveis para consolidar.", "merge_progress": "Consolidando… isso pode levar alguns minutos. Você pode fechar esta janela.", + "merge_run_aborted_toast": "A conexão com o nó foi redefinida, então a consolidação parou. Execute-a novamente para continuar.", + "merge_run_busy": "Uma consolidação está em andamento. Aguarde que termine ou interrompa-a primeiro.", + "merge_run_chip_fmt": "Consolidando %d/%d", + "merge_run_err_timeout": "O nó não respondeu a tempo; o lote ainda pode estar na fila.", + "merge_run_halted_fmt": "Consolidação interrompida após falhas repetidas: %s", + "merge_run_migration_pending": "Conclua ou cancele a migração da semente antes de consolidar.", + "merge_run_progress_fmt": "Consolidando… lote %d de cerca de %d, %d notas mescladas até agora. Você pode fechar esta janela; a operação continua em segundo plano.", + "merge_run_start_failed": "Não foi possível iniciar a consolidação — o nó não está conectado.", + "merge_run_status_not_run": "%s — não consolidado (%d notas)", + "merge_run_status_waiting": "%s — aguardando (%d notas)", + "merge_run_stop": "Parar", + "merge_run_stopped_toast": "Consolidação interrompida. Execute-a novamente para mesclar o restante.", + "merge_run_stopping": "Parando após os lotes já enviados…", + "merge_run_wait_batch": "Espaçando os lotes para que o nó continue respondendo…", + "merge_run_wait_send": "Pausado até que sua transação pendente seja concluída.", + "merge_run_wait_sync": "Pausado: o nó ainda está sincronizando.", "merge_scope_failed_fmt": "Não foi possível ler suas entradas: %s", "merge_scope_fmt": "%d entradas transparentes + %d blindadas · ~%s DRGX disponíveis", "merge_scope_loading": "Verificando suas entradas…", diff --git a/res/lang/ru.json b/res/lang/ru.json index d1345c4..dd56d38 100644 --- a/res/lang/ru.json +++ b/res/lang/ru.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "Макс. входов за партию", "merge_no_spendable": "Пока нет доступных для расходования входов для объединения.", "merge_progress": "Объединение… это может занять несколько минут. Вы можете закрыть это окно.", + "merge_run_aborted_toast": "Соединение с узлом было сброшено, поэтому объединение остановлено. Запустите его снова, чтобы продолжить.", + "merge_run_busy": "Идёт объединение. Дождитесь его завершения или сначала остановите его.", + "merge_run_chip_fmt": "Объединение %d/%d", + "merge_run_err_timeout": "Узел не ответил вовремя; партия, возможно, всё ещё в очереди.", + "merge_run_halted_fmt": "Объединение остановлено после повторных сбоев: %s", + "merge_run_migration_pending": "Завершите или отмените перенос сид-фразы перед объединением.", + "merge_run_progress_fmt": "Объединение… партия %d из примерно %d, объединено заметок: %d. Это окно можно закрыть — операция продолжится в фоне.", + "merge_run_start_failed": "Не удалось начать объединение — узел не подключён.", + "merge_run_status_not_run": "%s — не объединено (%d заметок)", + "merge_run_status_waiting": "%s — ожидание (%d заметок)", + "merge_run_stop": "Остановить", + "merge_run_stopped_toast": "Объединение остановлено. Запустите его снова, чтобы объединить остальное.", + "merge_run_stopping": "Остановка после уже отправленных партий…", + "merge_run_wait_batch": "Партии отправляются с паузами, чтобы узел оставался отзывчивым…", + "merge_run_wait_send": "Приостановлено до завершения вашей ожидающей транзакции.", + "merge_run_wait_sync": "Приостановлено: узел ещё синхронизируется.", "merge_scope_failed_fmt": "Не удалось прочитать ваши входы: %s", "merge_scope_fmt": "%d прозрачных + %d защищённых входов · ~%s DRGX доступно", "merge_scope_loading": "Проверка ваших входов…", diff --git a/res/lang/zh.json b/res/lang/zh.json index 8780511..f3b7704 100644 --- a/res/lang/zh.json +++ b/res/lang/zh.json @@ -1183,6 +1183,22 @@ "merge_max_inputs": "每批最大输入数", "merge_no_spendable": "暂无可合并的可用输入。", "merge_progress": "正在合并…这可能需要几分钟。您可以关闭此窗口。", + "merge_run_aborted_toast": "节点连接已重置，合并已停止。再次运行以继续。", + "merge_run_busy": "正在进行合并。请等待其完成，或先将其停止。", + "merge_run_chip_fmt": "合并中 %d/%d", + "merge_run_err_timeout": "节点未及时响应；该批次可能仍在队列中。", + "merge_run_halted_fmt": "多次失败后合并已停止：%s", + "merge_run_migration_pending": "请先完成或取消助记词迁移，再进行合并。", + "merge_run_progress_fmt": "正在合并… 第 %d 批（共约 %d 批），已合并 %d 张票据。可以关闭此窗口，合并会在后台继续。", + "merge_run_start_failed": "无法开始合并——节点未连接。", + "merge_run_status_not_run": "%s——未合并（%d 张票据）", + "merge_run_status_waiting": "%s——等待中（%d 张票据）", + "merge_run_stop": "停止", + "merge_run_stopped_toast": "合并已停止。再次运行以合并其余部分。", + "merge_run_stopping": "已发送的批次完成后停止…", + "merge_run_wait_batch": "正在间隔各批次，以保持节点响应…", + "merge_run_wait_send": "已暂停，等待您的待处理交易完成。", + "merge_run_wait_sync": "已暂停：节点仍在同步。", "merge_scope_failed_fmt": "无法读取您的输入：%s", "merge_scope_fmt": "%d 笔透明输入 + %d 笔屏蔽输入 ・ 约 %s DRGX 可用", "merge_scope_loading": "正在检查您的输入…", diff --git a/res/themes/ui.toml b/res/themes/ui.toml index f9cc512..6d49764 100644 --- a/res/themes/ui.toml +++ b/res/themes/ui.toml @@ -1097,6 +1097,7 @@ fee-input = { width = 150 } utxo-limit-input = { width = 150 } shield-button = { width = 120, font = "button" } cancel-button = { width = 100, font = "button" } +list-max-rows = { size = 7 } # address picker / run rows shown before the list scrolls [dialogs.request-payment] width = 500.0 diff --git a/src/app.cpp b/src/app.cpp index 7458d13..d6bb41c 100644 --- a/src/app.cpp +++ b/src/app.cpp @@ -1427,6 +1427,8 @@ void App::update() // While the address scan that normally ends the send progress can't run (the background, a catch-up, the // connect gate, a retry hold), end it once the operations are done: it keeps the window rendering. if (send_progress_active_ && !addressScanCanRunNow()) maybeFinishTransactionSendProgress(); + // C210: the "Keep in each address" consolidation run, if one is going (paced; one RPC at a time). + pumpConsolidation(); // After a bootstrap the preserved wallet.dat has stale spent-state and must be reconciled // against the freshly-imported chain, or the first send tries to spend already-spent notes and @@ -2979,6 +2981,30 @@ void App::renderStatusBar() ImGui::TextColored(ImVec4(0.6f, 0.8f, 1.0f, 1.0f), "%s%s", TR("sb_importing_keys"), dotStr); } + // C210: a consolidation run keeps going after its dialog closes. A static chip (no animation, no frames of + // its own); clicking it reopens the run view. + if (consolidation_.active()) { + const services::RunProgress p = consolidation_.progress(); + const ImVec4 col(0.6f, 0.8f, 1.0f, 1.0f); + ImGui::SameLine(0, sbSectionGap); + ImGui::TextDisabled("|"); + ImGui::SameLine(0, sbSeparatorGap); + ImGui::BeginGroup(); + ImGui::PushFont(ui::material::Type().iconSmall()); + ImGui::TextColored(col, ICON_MD_CALL_MERGE); + ImGui::PopFont(); + ImGui::SameLine(0, sbIconTextGap); + ImGui::TextColored(col, TR("merge_run_chip_fmt"), p.batchesSubmitted, std::max(1, p.batchesEstimated)); + ImGui::EndGroup(); + if (ImGui::IsItemHovered()) { + ImGui::SetMouseCursor(ImGuiMouseCursor_Hand); + if (!p.stopping) { + if (const char* k = services::runWaitKey(p.wait)) ui::material::Tooltip("%s", TR(k)); + } + } + if (ImGui::IsItemClicked()) ui::ShieldDialog::showConsolidate(); // reopens the run view + } + // Right side: version always at far right, connection status to its left. // Compute positions dynamically from actual text widths so they // never overlap and always stay within the window at any font scale. diff --git a/src/app.h b/src/app.h index c2c4b35..08347a4 100644 --- a/src/app.h +++ b/src/app.h @@ -31,6 +31,7 @@ #include "services/scan_policy.h" #include "services/wallet_scan_governor.h" #include "services/chat_scan_policy.h" +#include "services/note_consolidation.h" #include "services/network_refresh_service.h" #include "services/wallet_security_controller.h" #include "services/wallet_security_workflow.h" @@ -492,6 +493,19 @@ public: // full node that is past warmup, not rescanning and not busy. bool walletScanReady() const; + // C210: the "Keep in each address" consolidation run. Owned here so it keeps going when the Merge dialog + // closes; pumped from update(). Full node only (refused in Lite). Main thread. + const services::ConsolidationRun& consolidation() const { return consolidation_; } + bool startConsolidation(const std::vector& groups, double fee, int batchLimit); + void stopConsolidation() { consolidation_.requestStop(); } + void dismissConsolidation() { consolidation_.dismiss(); } // no-op while running + // A one-off wallet-wide scan the user asked for (the Merge dialog's scope read, a merge): never held back, + // but charged to the scan budget (ScanStream::Adhoc) so the routine scans make room. Main thread. + services::ScanTicket beginAdhocWalletScan(); + void endAdhocWalletScan(const services::ScanTicket& ticket, double costMs, services::ScanOutcome outcome); + // A GUI wallet scan is in the daemon or queued for it: a one-off scan waits rather than stack on cs_main. + bool walletScanInFlight() const { return scan_governor_.anyInFlight(); } + // Force refresh void refreshNow(); // Called on window restore: drop any daemon-output backlog (update() runs while minimized, but its @@ -1661,6 +1675,13 @@ private: std::atomic opid_filter_unsupported_{false}; // see opidFilterUnsupported(); set on the worker std::atomic maint_calls_{0}; // maintenance calls in flight (MaintenanceCallScope) std::atomic maint_abort_{false}; // set by abortMaintenanceCalls(), cleared on connect + services::ConsolidationRun consolidation_; // C210 (see consolidation()) + std::uint64_t rpc_session_ = 0; // ++ in onConnected/onDisconnected: an orphan merge is tracked only within one + double run_lock_resync_at_ = 0.0; // steadySeconds() before which the run asks for no lock-state re-read + void pumpConsolidation(); + void postConsolidationCall(int addrIdx); + void postConsolidationPoll(); + void finishConsolidation(); // Consecutive Core-refresh cycles where BOTH core RPCs failed → likely a dead // connection. After kCoreFailuresBeforeDisconnect, tear down and reconnect. int consecutive_core_failures_ = 0; diff --git a/src/app_network.cpp b/src/app_network.cpp index f4dd3a0..c8eb6d1 100644 --- a/src/app_network.cpp +++ b/src/app_network.cpp @@ -690,6 +690,8 @@ void App::tryConnect() void App::onConnected() { maint_abort_.store(false); // maintenance calls queued before this connect have already failed (FIFO) + ++rpc_session_; + if (consolidation_.abort()) ui::Notifications::instance().warning(TR("merge_run_aborted_toast")); state_.connected = true; ui::ShieldDialog::invalidateScope(); // inputs read before this connect may be another wallet's const bool completedSwitch = wallet_switch_pending_confirm_.exchange(false); // this connect confirms a switch @@ -873,6 +875,8 @@ void App::onDisconnected(const std::string& reason) first_balance_read_tried_ = false; address_owed_next_try_ = {}; mempool_known_ = false; + ++rpc_session_; + if (consolidation_.abort()) ui::Notifications::instance().warning(TR("merge_run_aborted_toast")); ui::ShieldDialog::invalidateScope(); address_validation_cache_dirty_ = true; resetTransactionHistoryCacheSession(); @@ -1352,6 +1356,17 @@ void App::markWalletMutated(bool forceBalance) network_refresh_.markWalletMutationRefresh(); // Core / Transactions / Addresses due now } +services::ScanTicket App::beginAdhocWalletScan() +{ + return scan_governor_.beginExplicit(services::ScanStream::Adhoc, currentScanSignals(), scanClockNow(), + /*servesRoutine=*/false); +} + +void App::endAdhocWalletScan(const services::ScanTicket& ticket, double costMs, services::ScanOutcome outcome) +{ + scan_governor_.complete(ticket, costMs, outcome, scanClockNow()); +} + bool App::invokeSendResultCallback(const std::string& opid, bool ok, const std::string& result) { @@ -1511,6 +1526,10 @@ void App::switchToWallet(const std::string& walletFile, bool stopDaemonConfirmed ui::Notifications::instance().warning("Finish or cancel the seed migration before switching wallets."); return; } + if (consolidation_.active()) { + ui::Notifications::instance().warning(TR("merge_run_busy")); + return; + } if (!isUsingEmbeddedDaemon()) { ui::Notifications::instance().warning("Switching wallets needs the embedded daemon — stop any external dragonxd first."); return; @@ -5468,6 +5487,167 @@ void App::maybeRemindSeedBackup() // shrinks in place and shielded-note witness data accumulates, so a mining/shielded wallet can grow // unbounded. Fires ONCE (persisted flag) a warning toast + a clickable "Consolidate notes…" entry in the // bell/alert panel that opens Merge to Address; re-arms if the file later drops back under the threshold. +bool App::startConsolidation(const std::vector& groups, double fee, int batchLimit) +{ + auto& n = ui::Notifications::instance(); + if (isLiteBuild() || capture_mode_ || shutting_down_.load()) return false; + if (consolidation_.active()) { + n.warning(TR("merge_run_busy")); + return false; + } + if (show_seed_migration_ || (settings_ && settings_->getSeedMigrationPending())) { + n.warning(TR("merge_run_migration_pending")); + return false; + } + if (!rpc_ || !rpc_->isConnected() || !worker_ || !state_.connected || isWalletSwitchInProgress()) { + n.error(TR("merge_run_start_failed")); + return false; + } + services::RunConfig cfg; + cfg.fee = fee; + cfg.limit = batchLimit; + if (!consolidation_.start(groups, cfg, services::steadySeconds())) { + n.warning(TR("merge_keep_each_select")); + return false; + } + n.success(TR("merge_started")); + return true; // syncing, the lock, a busy node and pending sends are the run's own pauses +} + +void App::pumpConsolidation() +{ + if (!consolidation_.active()) return; + const double now = services::steadySeconds(); + const services::ScanContext ctx = scanContext(); + const bool unlockExpired = state_.encrypted && !state_.locked && state_.unlocked_until > 0 && + std::time(nullptr) >= static_cast(state_.unlocked_until); + services::RunEnv env; + // Not walletScanReady(): the run is the user's job (no background hold, no connect gate), and it tells its + // pauses apart. Not rpc_->isWarmingUp(): that client flag sticks for the session. + env.nodeUsable = !isLiteBuild() && !shutting_down_.load() && !daemon_restarting_.load() && worker_ && + ctx.connected && !ctx.warmingUp && !ctx.runtimeRescan && !state_.sync.rescanning; + env.busy = ctx.rpcBusy; + env.callsAllowed = !ctx.syncing; + env.locked = ctx.locked || unlockExpired; + env.userSendBusy = hasTransactionSendProgress(); // a send / shield / auto-shield / chat op of ours goes first + env.scanBusy = scan_governor_.anyInFlight(); // never stack a merge scan on another GUI scan (both lanes) + if ((consolidation_.consumeLockResyncRequest() || unlockExpired) && now >= run_lock_resync_at_) { + run_lock_resync_at_ = now + 30.0; + refreshEncryptionState(); + } + const services::RunStep step = consolidation_.next(env, now); + if (step.action == services::RunAction::Call) postConsolidationCall(step.addrIdx); + else if (step.action == services::RunAction::Poll) postConsolidationPoll(); + else if (step.action == services::RunAction::Finish) finishConsolidation(); +} + +void App::postConsolidationCall(int addrIdx) +{ + if (!worker_ || !rpc_) return; // before markCallPosted: asked again next update + const auto& addrs = consolidation_.addresses(); + if (addrIdx < 0 || addrIdx >= static_cast(addrs.size())) return; + const std::string address = addrs[addrIdx].group.address; + const std::uint64_t gen = consolidation_.gen(); + const std::uint64_t session = rpc_session_; + const double fee = consolidation_.config().fee; + const int limit = consolidation_.config().limit; + const services::ScanTicket ticket = beginAdhocWalletScan(); // GetFilteredNotes: a whole-wallet scan + consolidation_.markCallPosted(addrIdx, services::steadySeconds()); + worker_->post([this, rpc = rpc_.get(), address, addrIdx, gen, fee, limit, session, ticket]() -> rpc::RPCWorker::MainCb { + services::MergeCallResult r; + rpc::RpcFailureKind kind = rpc::RpcFailureKind::None; + const auto t0 = std::chrono::steady_clock::now(); + try { + MaintenanceCallScope maint(*this); // a disconnect aborts it, or refuses it unsent + rpc::RPCClient::TraceScope trace("Consolidate / Merge batch"); + // fromaddresses = [addr], toaddress = addr: the address's notes merge back into it. + const nlohmann::json res = rpc->call("z_mergetoaddress", + {nlohmann::json::array({address}), address, fee, limit, limit}, services::kRunCallTimeoutSec); + services::readMergeCallResult(res, r); + } catch (const std::exception& e) { + r = services::MergeCallResult{}; + r.error = e.what(); + if (r.error.empty()) r.error = "error"; + kind = services::mergeCallFailureKind(e); // sent but cut off: uncertain, never resent + } catch (...) { + r = services::MergeCallResult{}; + r.error = "unknown error"; + kind = rpc::RpcFailureKind::Other; + } + const double secs = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + return [this, gen, addrIdx, r, kind, secs, session, ticket]() { + endAdhocWalletScan(ticket, secs * 1000.0, services::chatScanOutcome(r.error.empty(), kind)); + const std::string orphan = consolidation_.onCallResult(gen, addrIdx, r, kind, secs, services::steadySeconds()); + if (!orphan.empty() && session == rpc_session_) trackOperation(orphan); // never lose an accepted merge + }; + }); +} + +void App::postConsolidationPoll() +{ + if (!worker_ || !rpc_) return; + const std::vector ids = consolidation_.outstandingOpids(); + const auto filter = NetworkRefreshService::buildOperationFilterParams(ids); + if (!filter) return; + const std::set wanted(ids.begin(), ids.end()); + const std::uint64_t gen = consolidation_.gen(); + consolidation_.markPollPosted(services::steadySeconds()); + // Not charged: z_getoperationstatus reads only the async ops map. It needs the filter fallback itself: the + // shared opid poller only learns that the filter is unsupported when it has ops of its own. + worker_->post([this, rpc = rpc_.get(), params = *filter, wanted, gen]() -> rpc::RPCWorker::MainCb { + bool ok = false; + std::vector reports; + try { + rpc::RPCClient::TraceScope trace("Consolidate / Batch status"); + nlohmann::json result; + if (!opid_filter_unsupported_.load(std::memory_order_relaxed)) { + try { + result = rpc->call("z_getoperationstatus", params); + } catch (const rpc::RpcError& e) { + if (e.kind() == rpc::RpcFailureKind::Warmup) throw; + result = rpc->call("z_getoperationstatus", nlohmann::json::array()); + opid_filter_unsupported_.store(true, std::memory_order_relaxed); + } + } else { + result = rpc->call("z_getoperationstatus", nlohmann::json::array()); + } + ok = result.is_array(); + reports = services::parseOpReports(result, wanted); + } catch (...) { + ok = false; + reports.clear(); + } + return [this, gen, ok, reports = std::move(reports)]() { + consolidation_.onPollResult(gen, ok, reports, services::steadySeconds()); + }; + }); +} + +void App::finishConsolidation() +{ + const services::RunSummary s = consolidation_.summary(); + // 2E: a batch of user ops re-reads the wallet and forces one balance, at its end. + if (s.batchesSucceeded > 0) markWalletMutated(/*forceBalance=*/true); + else if (s.batchesSubmitted > 0) noteWalletMutation(); + std::string err = s.firstError.empty() ? std::string(TR("shield_unknown_error")) : s.firstError; + if (err.size() > 160) err = err.substr(0, 160) + "..."; + auto& n = ui::Notifications::instance(); + switch (services::runVerdict(s)) { + case services::RunVerdict::Clean: n.success(TR("shield_merge_done")); break; + case services::RunVerdict::NothingSubmitted: n.error(TR("merge_keep_each_nothing_submitted")); break; + case services::RunVerdict::SomeFailed: n.error(std::string(TR("shield_op_failed")) + err); break; + case services::RunVerdict::Unconfirmed: n.warning(TR("send_status_unconfirmed")); break; + case services::RunVerdict::Stopped: n.info(TR("merge_run_stopped_toast")); break; + case services::RunVerdict::Halted: { + char b[512]; + std::snprintf(b, sizeof(b), TR("merge_run_halted_fmt"), err.c_str()); + n.error(b); + break; + } + default: break; // Aborted is announced where it happens; it never reaches Finish + } +} + void App::maybeWarnLargeWallet() { if (capture_mode_ || lite_wallet_) return; // no live nags during a UI sweep; lite has no wallet.dat @@ -5660,6 +5840,10 @@ void App::beginCreateSeedWallet() void App::showSeedMigrationDialog() { + if (consolidation_.active()) { + ui::Notifications::instance().warning(TR("merge_run_busy")); + return; + } show_seed_migration_ = true; // Resume a pending migration. If a sweep was already submitted (txid persisted), resume at the // confirm/adopt stage — re-derived from the chain — rather than sweeping again; otherwise start @@ -5765,6 +5949,10 @@ void App::refreshSeedMigrationBalance() // huge-input wallet may leave a small remainder in the (preserved) legacy backup. void App::beginSweepToSeedWallet() { + if (consolidation_.active()) { + ui::Notifications::instance().warning(TR("merge_run_busy")); + return; + } if (!rpc_ || !worker_ || !state_.connected) { seed_migration_status_ = "Not connected to the daemon."; seed_migration_step_ = SeedMigrationStep::Error; return; diff --git a/src/services/note_consolidation.cpp b/src/services/note_consolidation.cpp index 6a6caa3..5fd4b53 100644 --- a/src/services/note_consolidation.cpp +++ b/src/services/note_consolidation.cpp @@ -6,7 +6,9 @@ #include #include +#include #include +#include namespace dragonx { namespace services { @@ -58,24 +60,489 @@ int estimateBatches(int notes, int limit) return notes / limit + ((notes % limit) >= 2 ? 1 : 0); } +bool recordMergeResult(AddressRun& run, const MergeCallResult& r, int maxBatches) +{ + if (!r.error.empty()) { + run.error = r.error; // earlier batches (if any) are already submitted and stand + return false; + } + run.opids.push_back(r.opid); + run.notesMerged += r.merging; + run.remaining = r.remaining; + // One leftover note can't be merged into its own address. + return r.remaining >= 2 && static_cast(run.opids.size()) < maxBatches; +} + AddressRun consolidateInPlace(const std::string& address, const MergeCall& call, int maxBatches) { AddressRun run; run.address = address; - for (int b = 0; b < maxBatches; ++b) { - MergeCallResult r = call(address); - if (!r.error.empty()) { - run.error = r.error; // earlier batches (if any) are already submitted and stand - break; - } - run.opids.push_back(r.opid); - run.notesMerged += r.merging; - run.remaining = r.remaining; - if (r.remaining < 2) break; // one leftover note can't be merged into its own address - } + if (maxBatches <= 0) return run; + while (recordMergeResult(run, call(address), maxBatches)) {} return run; } +namespace { +std::string lower(std::string s) +{ + for (auto& c : s) c = static_cast(std::tolower(static_cast(c))); + return s; +} +bool has(const std::string& haystack, const char* needle) { return haystack.find(needle) != std::string::npos; } + +int clampedCount(const nlohmann::json& obj, const char* key) +{ + const auto it = obj.find(key); + if (it == obj.end()) return 0; + if (it->is_number_unsigned()) { + const auto v = it->get(); + return v > static_cast(INT_MAX) ? INT_MAX : static_cast(v); + } + if (it->is_number_integer()) { + const auto v = it->get(); + return v < 0 ? 0 : (v > INT_MAX ? INT_MAX : static_cast(v)); + } + return 0; +} + +std::string stringField(const nlohmann::json& obj, const char* key) +{ + const auto it = obj.find(key); + return (it != obj.end() && it->is_string()) ? it->get() : std::string(); +} +} // namespace + +void readMergeCallResult(const nlohmann::json& res, MergeCallResult& out) +{ + out = MergeCallResult{}; + if (res.is_object()) { + out.opid = stringField(res, "opid"); + out.merging = clampedCount(res, "mergingNotes"); + out.remaining = clampedCount(res, "remainingNotes"); + } + if (out.opid.empty()) { + out = MergeCallResult{}; + out.error = "no operation id"; + } +} + +OpState opStateFromStatus(const std::string& status) +{ + if (status == "queued") return OpState::Queued; + if (status == "executing") return OpState::Executing; + if (status == "success") return OpState::Success; + if (status == "failed" || status == "cancelled" || status == "canceled") return OpState::Failed; + return OpState::Unknown; +} + +std::vector parseOpReports(const nlohmann::json& statusArray, const std::set& wanted) +{ + std::vector out; + if (!statusArray.is_array()) return out; + for (const auto& op : statusArray) { + if (!op.is_object()) continue; + OpReport rep; + rep.id = stringField(op, "id"); + if (rep.id.empty()) continue; + if (!wanted.empty() && !wanted.count(rep.id)) continue; + rep.status = stringField(op, "status"); + const auto err = op.find("error"); + if (err != op.end() && err->is_object()) rep.error = stringField(*err, "message"); + const auto res = op.find("result"); + if (res != op.end() && res->is_object()) rep.txid = stringField(*res, "txid"); + out.push_back(std::move(rep)); + } + return out; +} + +MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& message) +{ + using K = rpc::RpcFailureKind; + switch (kind) { + case K::None: return MergeFailure::Uncertain; // answered without an opid: it may have queued the merge + case K::Deferred: + case K::NotListening: + case K::ConnectTimeout: + case K::Warmup: + case K::QueueFull: return MergeFailure::NotSent; // turned away before z_mergetoaddress ran + case K::ReplyTimeout: // the daemon keeps running it and may queue the merge + case K::Aborted: // our abort: if sent it may have queued; if not, the session is ending (result stale) + return MergeFailure::Uncertain; + case K::ShuttingDown: + case K::Unhandled: return MergeFailure::Transient; // the HTTP layer dropped it + case K::Auth: return MergeFailure::Fatal; + default: break; // AppError, Other: by message + } + const std::string m = lower(message); + if (has(m, "not connected")) return MergeFailure::NotSent; // RPCClient throws before sending + if (has(m, "still syncing") || has(m, "fully synced")) return MergeFailure::Syncing; // THROW_IF_SYNCING (cheap) + if (has(m, "spending key") || has(m, "wallet passphrase") || has(m, "walletpassphrase")) return MergeFailure::NeedsUnlock; + if (has(m, "queue is shutting down")) return MergeFailure::Transient; // "...; the operation was not queued" + if (has(m, "could not find any funds") || has(m, "already merged") || has(m, "insufficient funds") || + has(m, "greater than the net amount")) + return MergeFailure::NothingLeft; + return MergeFailure::Fatal; +} + +rpc::RpcFailureKind mergeCallFailureKind(const std::exception& e) +{ + const rpc::RpcFailureKind kind = rpc::rpcFailureKindOf(e); + if (kind == rpc::RpcFailureKind::Other) { + if (const auto* t = dynamic_cast(&e); t && t->requestSent && t->httpStatus == 0) + return rpc::RpcFailureKind::ReplyTimeout; + } + return kind; +} + +RunVerdict runVerdict(const RunSummary& s) +{ + switch (s.end) { + case RunEnd::None: return RunVerdict::None; + case RunEnd::Stopped: return RunVerdict::Stopped; + case RunEnd::Halted: return RunVerdict::Halted; + case RunEnd::Aborted: return RunVerdict::Aborted; + case RunEnd::Completed: break; + } + if (s.batchesSubmitted == 0 && s.addressesUnconfirmed == 0) return RunVerdict::NothingSubmitted; + if (s.addressesFailed > 0) return RunVerdict::SomeFailed; + if (s.addressesUnconfirmed > 0) return RunVerdict::Unconfirmed; + return RunVerdict::Clean; +} + +const char* runWaitKey(RunWait wait) +{ + switch (wait) { + case RunWait::Node: + case RunWait::Busy: return "merge_waiting_node"; + case RunWait::Syncing: return "merge_run_wait_sync"; + case RunWait::Locked: return "merge_wait_locked"; + case RunWait::UserSend: return "merge_run_wait_send"; + case RunWait::Scan: + case RunWait::Batch: return "merge_run_wait_batch"; + case RunWait::None: + case RunWait::InFlight: break; + } + return nullptr; +} + +bool ConsolidationRun::start(const std::vector& groups, RunConfig cfg, double now) +{ + if (phase_ == Phase::Running) return false; + cfg.limit = std::max(2, cfg.limit); + cfg.maxBatches = std::max(1, cfg.maxBatches); + std::vector addrs; + for (const auto& g : groups) { + if (g.address.empty() || !worthConsolidating(g, cfg.fee)) continue; + RunAddress a; + a.group = g; + a.run.address = g.address; + addrs.push_back(std::move(a)); + } + if (addrs.empty()) return false; + const std::uint64_t gen = gen_ + 1; + *this = ConsolidationRun(); + gen_ = gen; + cfg_ = cfg; + addrs_ = std::move(addrs); + phase_ = Phase::Running; + nextCallAt_ = nextPollAt_ = now; + lastPumpAt_ = now; + return true; +} + +void ConsolidationRun::requestStop() +{ + if (phase_ == Phase::Running) stopping_ = true; +} + +void ConsolidationRun::dismiss() +{ + if (phase_ == Phase::Running) return; + const std::uint64_t gen = gen_; // a stale result can never match a later run + *this = ConsolidationRun(); + gen_ = gen; +} + +bool ConsolidationRun::abort() +{ + if (phase_ != Phase::Running) return false; + ++gen_; + if (callInFlight_ && callAddr_ >= 0 && callAddr_ < static_cast(addrs_.size())) { + auto& a = addrs_[callAddr_]; + a.done = true; + worsen(a, AddrOutcome::Uncertain); + } + callInFlight_ = pollInFlight_ = false; + for (auto& b : batches_) { + if (!opStateTerminal(b.state)) applyOpState(b, OpState::Lost, nullptr); // "could not be confirmed" + } + phase_ = Phase::Finished; + end_ = RunEnd::Aborted; + lastWait_ = RunWait::None; + return true; +} + +void ConsolidationRun::endAddress(RunAddress& a, AddrOutcome o, const std::string& msg) +{ + a.done = true; + worsen(a, o); + if (a.run.error.empty()) a.run.error = msg; + if (a.run.opids.empty() && ++deadStreak_ >= kRunMaxFailStreak) halt(msg); +} + +void ConsolidationRun::applyOpState(RunBatch& b, OpState st, const OpReport* rep) +{ + b.state = st; + if (b.addrIdx < 0 || b.addrIdx >= static_cast(addrs_.size())) return; + RunAddress& a = addrs_[b.addrIdx]; + switch (st) { + case OpState::Success: + if (rep) b.txid = rep->txid; + ++a.opsDone; + opFailStreak_ = 0; + break; + case OpState::Failed: + if (rep) b.error = rep->error; + if (a.failure.empty()) a.failure = b.error; + a.done = true; + worsen(a, AddrOutcome::Failed); + if (++opFailStreak_ >= kRunMaxFailStreak) halt(b.error); + break; + case OpState::Lost: + a.done = true; + worsen(a, AddrOutcome::Uncertain); + break; + default: + break; + } +} + +RunStep ConsolidationRun::next(const RunEnv& env, double now) +{ + if (phase_ != Phase::Running) return {}; + if (callInFlight_ && now - callPostedAt_ > kRunWatchdogSec) { // its result never came: a late reply is an orphan + callInFlight_ = false; + ++gen_; + if (callAddr_ >= 0 && callAddr_ < static_cast(addrs_.size())) + endAddress(addrs_[callAddr_], AddrOutcome::Uncertain, ""); + nextCallAt_ = std::max(nextCallAt_, now + kRunMaxGapSec); + } + if (pollInFlight_ && now - pollPostedAt_ > kRunWatchdogSec) { + pollInFlight_ = false; + ++gen_; + } + // An op the daemon never reports done is presumed lost after an hour in which it could have been polled + // (not while the node was gone or busy: it may simply not have been asked), or after 4 h whatever happened. + const double dt = std::max(0.0, now - lastPumpAt_); + lastPumpAt_ = now; + const bool pollable = env.nodeUsable && !env.busy; + for (auto& b : batches_) { + if (opStateTerminal(b.state)) continue; + if (pollable) b.pollableSec += std::min(dt, std::max(0.0, now - b.acceptedAt)); + if (b.pollableSec > kRunOpMaxSec || now - b.acceptedAt > kRunOpHardMaxSec) applyOpState(b, OpState::Lost, nullptr); + } + while (cur_ < addrs_.size() && addrs_[cur_].done) ++cur_; + const bool moreCalls = !stopping_ && !halted_ && cur_ < addrs_.size(); + const int outstanding = static_cast(std::count_if(batches_.begin(), batches_.end(), + [](const RunBatch& b) { return !opStateTerminal(b.state); })); + auto wait = [&](RunWait w) { + lastWait_ = w; + return RunStep{}; + }; + if (!callInFlight_ && !pollInFlight_ && !moreCalls && outstanding == 0) { + phase_ = Phase::Finished; + end_ = halted_ ? RunEnd::Halted : (stopping_ && cur_ < addrs_.size()) ? RunEnd::Stopped : RunEnd::Completed; + lastWait_ = RunWait::None; + return {RunAction::Finish, -1}; // exactly once + } + if (callInFlight_ || pollInFlight_) return wait(RunWait::InFlight); // one run RPC at a time + if (!env.nodeUsable) return wait(RunWait::Node); // blocks polls too + if (env.busy) return wait(RunWait::Busy); // z_getoperationstatus needs cs_main + if (outstanding > 0 && now >= nextPollAt_) { + lastWait_ = RunWait::None; + return {RunAction::Poll, -1}; + } + if (!moreCalls) return wait(RunWait::None); // waiting for submitted batches + if (!env.callsAllowed || now < syncHoldUntil_) return wait(RunWait::Syncing); + if (env.locked || now < unlockHoldUntil_) return wait(RunWait::Locked); + if (env.userSendBusy) return wait(RunWait::UserSend); + if (now < nextCallAt_) return wait(RunWait::Batch); + if (!batches_.empty() && (batches_.back().state == OpState::Unknown || batches_.back().state == OpState::Queued)) + return wait(RunWait::Batch); // at most 2 unfinished ops, at most 2 x limit notes locked + if (env.scanBusy) return wait(RunWait::Scan); // another GUI wallet scan is in the daemon + lastWait_ = RunWait::None; + return {RunAction::Call, static_cast(cur_)}; +} + +void ConsolidationRun::markCallPosted(int addrIdx, double now) +{ + if (phase_ != Phase::Running || addrIdx < 0 || addrIdx >= static_cast(addrs_.size())) return; + callInFlight_ = true; + callAddr_ = addrIdx; + callPostedAt_ = now; + worsen(addrs_[addrIdx], AddrOutcome::Running); +} + +void ConsolidationRun::markPollPosted(double now) +{ + if (phase_ != Phase::Running) return; + pollInFlight_ = true; + pollPostedAt_ = now; +} + +std::string ConsolidationRun::onCallResult(std::uint64_t gen, int addrIdx, const MergeCallResult& r, + rpc::RpcFailureKind kind, double callSec, double now) +{ + const bool accepted = r.error.empty() && !r.opid.empty(); + if (gen != gen_ || phase_ != Phase::Running || !callInFlight_ || addrIdx != callAddr_) + return accepted ? r.opid : std::string(); + callInFlight_ = false; + const double gapAt = now + std::clamp(callSec, kRunMinGapSec, kRunMaxGapSec); + nextCallAt_ = gapAt; + RunAddress& a = addrs_[addrIdx]; + if (accepted) { + a.transient = 0; + deadStreak_ = 0; + const bool more = recordMergeResult(a.run, r, cfg_.maxBatches); + RunBatch b; + b.addrIdx = addrIdx; + b.opid = r.opid; + b.merging = r.merging; + b.acceptedAt = now; + batches_.push_back(std::move(b)); + nextPollAt_ = now; // learn its state before the next call + if (!more) { + a.done = true; + worsen(a, AddrOutcome::Done); + } + return {}; + } + const std::string msg = r.error.empty() ? std::string("no operation id") : r.error; + switch (classifyMergeFailure(kind, msg)) { + case MergeFailure::NotSent: // the same batch again, no retry budget + nextCallAt_ = std::max(gapAt, now + kRunNotSentRetrySec); + break; + case MergeFailure::Syncing: // no retry budget + syncHoldUntil_ = now + kRunSyncHoldSec; + break; + case MergeFailure::NeedsUnlock: + if (++a.unlockRefusals >= kRunMaxUnlockRefusals) { + endAddress(a, AddrOutcome::Failed, msg); + } else { + unlockHoldUntil_ = now + kRunUnlockHoldSec; + lockResync_ = true; + } + break; + case MergeFailure::NothingLeft: + if (a.run.opids.empty()) { + endAddress(a, AddrOutcome::Failed, msg); + } else { + a.done = true; + worsen(a, AddrOutcome::Done); + } + break; + case MergeFailure::Uncertain: // never resent + endAddress(a, AddrOutcome::Uncertain, msg); + break; + case MergeFailure::Transient: + if (++a.transient > kRunMaxTransientRetries) { + endAddress(a, AddrOutcome::Failed, msg); + } else { + nextCallAt_ = std::max(gapAt, now + std::min(60.0, 5.0 * static_cast(1 << (a.transient - 1)))); + } + break; + case MergeFailure::Fatal: + endAddress(a, AddrOutcome::Failed, msg); + break; + } + return {}; +} + +void ConsolidationRun::onPollResult(std::uint64_t gen, bool ok, const std::vector& reports, double now) +{ + if (gen != gen_ || phase_ != Phase::Running || !pollInFlight_) return; + pollInFlight_ = false; + nextPollAt_ = now + kRunPollIntervalSec; + if (!ok) return; + // No call can be in flight during a poll, so the non-terminal batches now are exactly the ids it asked about. + std::map byId; + for (const auto& rep : reports) byId[rep.id] = &rep; + for (auto& b : batches_) { + if (opStateTerminal(b.state)) continue; + const auto it = byId.find(b.opid); + if (it == byId.end()) { + if (++b.misses >= kRunMaxPollMisses) applyOpState(b, OpState::Lost, nullptr); + continue; + } + b.misses = 0; + const OpState st = opStateFromStatus(it->second->status); + if (st != OpState::Unknown) applyOpState(b, st, it->second); + } +} + +bool ConsolidationRun::consumeLockResyncRequest() +{ + const bool r = lockResync_; + lockResync_ = false; + return r; +} + +std::vector ConsolidationRun::outstandingOpids() const +{ + std::vector ids; + for (const auto& b : batches_) { + if (!opStateTerminal(b.state)) ids.push_back(b.opid); + } + return ids; +} + +RunProgress ConsolidationRun::progress() const +{ + RunProgress p; + p.addresses = static_cast(addrs_.size()); + for (const auto& a : addrs_) { + if (a.done) ++p.addressesDone; + const int submitted = static_cast(a.run.opids.size()); + p.batchesEstimated += a.done ? submitted + : std::max(std::min(estimateBatches(a.group.notes, cfg_.limit), cfg_.maxBatches), submitted); + } + p.batchesSubmitted = static_cast(batches_.size()); + for (const auto& b : batches_) { + if (opStateTerminal(b.state)) ++p.batchesDone; + if (b.state == OpState::Success) p.notesMerged += b.merging; + } + p.stopping = stopping_; + p.wait = lastWait_; + return p; +} + +RunSummary ConsolidationRun::summary() const +{ + RunSummary s; + s.end = end_; + s.addresses = static_cast(addrs_.size()); + for (const auto& a : addrs_) { + if (a.outcome == AddrOutcome::Failed) ++s.addressesFailed; + if (a.outcome == AddrOutcome::Uncertain) ++s.addressesUnconfirmed; + } + s.batchesSubmitted = static_cast(batches_.size()); + for (const auto& b : batches_) { + if (b.state == OpState::Success) { + ++s.batchesSucceeded; + s.notesMerged += b.merging; + } + } + s.firstError = haltError_; + if (s.firstError.empty()) { + for (const auto& a : addrs_) { + if (a.outcome != AddrOutcome::Failed) continue; + s.firstError = !a.failure.empty() ? a.failure : a.run.error; + if (!s.firstError.empty()) break; + } + } + return s; +} + ScopeWant scopeScanWant(const ScopeScan& scan, const ScopeInputs& in) { if (!in.mergeMode || scan.phase == ScopePhase::Loaded) return ScopeWant::Nothing; diff --git a/src/services/note_consolidation.h b/src/services/note_consolidation.h index 06b6a49..43775eb 100644 --- a/src/services/note_consolidation.h +++ b/src/services/note_consolidation.h @@ -17,6 +17,7 @@ #include #include #include +#include #include #include @@ -67,6 +68,168 @@ struct AddressRun { // locks a batch's notes before the call returns, so each call takes the next batch. At most maxBatches. AddressRun consolidateInPlace(const std::string& address, const MergeCall& call, int maxBatches = 20); +// ── Merge-call stepper ── +// One call's result into the address's run (r.error set: run.error = r.error, false). True while another call +// for this address is worth making: at least 2 unlocked notes left and fewer than maxBatches submitted. +bool recordMergeResult(AddressRun& run, const MergeCallResult& r, int maxBatches = 20); +// z_mergetoaddress's reply. Never throws. A non-empty string "opid" = accepted; mergingNotes/remainingNotes only +// when integral (clamped to [0, INT_MAX]), else 0; no opid: error "no operation id". +void readMergeCallResult(const nlohmann::json& res, MergeCallResult& out); + +// ── Async op status ── +enum class OpState : std::uint8_t { Unknown, Queued, Executing, Success, Failed, Lost }; +inline bool opStateTerminal(OpState s) { return s == OpState::Success || s == OpState::Failed || s == OpState::Lost; } +struct OpReport { + std::string id, status, error, txid; // error = error.message, txid = result.txid +}; +OpState opStateFromStatus(const std::string& status); // queued/executing/success; failed|cancelled|canceled: Failed +// Type-checked, never throws. Skips non-objects and missing/non-string/empty ids; keeps only `wanted` (all if empty). +std::vector parseOpReports(const nlohmann::json& statusArray, const std::set& wanted = {}); + +// ── Merge-call failures ── +enum class MergeFailure : std::uint8_t { + NotSent, // turned away before z_mergetoaddress ran: the same batch again, later + Syncing, // THROW_IF_SYNCING: hold the run while the node catches up + NeedsUnlock, // a locked keystore ("could not find spending key") + NothingLeft, // no funds / already merged / too small for the fee + Uncertain, // it may have queued the merge (a reply timeout, an answer without an opid): never resent + Transient, // the HTTP layer dropped it, or the async queue is shutting down: retry with backoff + Fatal, // anything else: this address is done +}; +MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& message); +// The failure kind of a z_mergetoaddress call that threw. Sent, then cut off before any HTTP status +// (connection reset, an empty or partial reply): the daemon may have queued the merge, exactly as after a +// reply timeout, so it reads as one. Any other failure keeps its kind. +rpc::RpcFailureKind mergeCallFailureKind(const std::exception& e); + +// ── Paced per-address consolidation run (owned by App, pumped from App::update(); main thread) ── +// The "Keep in each address" mode used to post one task that called z_mergetoaddress for every batch of every +// address back to back. Each call is a whole-wallet scan under cs_main, and the daemon runs the ops one at a +// time anyway. Here: one run RPC at a time, the next call only after the previous batch left the queue and a +// gap as long as the last call took, failures classified (nothing that may have queued is ever resent), and a +// breaker after three dead addresses or three failed ops in a row. +inline constexpr double kRunMinGapSec = 2.0; +inline constexpr double kRunMaxGapSec = 120.0; +inline constexpr double kRunPollIntervalSec = 3.0; +inline constexpr long kRunCallTimeoutSec = 120; +inline constexpr double kRunWatchdogSec = 1800.0; // a call or poll outstanding this long is presumed lost +inline constexpr double kRunOpMaxSec = 3600.0; // an op not terminal after this much pollable time is presumed lost +inline constexpr double kRunOpHardMaxSec = 4 * kRunOpMaxSec; // ...and after this long whatever happened +inline constexpr double kRunNotSentRetrySec = 5.0; +inline constexpr double kRunSyncHoldSec = 30.0; +inline constexpr double kRunUnlockHoldSec = 30.0; +inline constexpr int kRunMaxTransientRetries = 5; +inline constexpr int kRunMaxUnlockRefusals = 2; +inline constexpr int kRunMaxFailStreak = 3; +inline constexpr int kRunMaxPollMisses = 2; + +struct RunConfig { + double fee = 0.0001; + int limit = 50; + int maxBatches = 20; +}; +struct RunEnv { + bool nodeUsable = false; // connected, not warming up or rescanning + bool busy = false; // the busy gate holds RPCs that need cs_main + bool callsAllowed = true; // not syncing / settling + bool locked = false; + bool userSendBusy = false; // a send / shield / auto-shield / chat op of ours is in progress + bool scanBusy = false; // another GUI wallet scan is in the daemon +}; +enum class RunAction : std::uint8_t { None, Call, Poll, Finish }; +enum class RunWait : std::uint8_t { None, InFlight, Node, Busy, Syncing, Locked, UserSend, Scan, Batch }; +enum class RunEnd : std::uint8_t { None, Completed, Stopped, Halted, Aborted }; +enum class AddrOutcome : std::uint8_t { Pending, Running, Done, Uncertain, Failed }; // ordered: worse wins +enum class RunVerdict : std::uint8_t { None, Clean, NothingSubmitted, SomeFailed, Unconfirmed, Stopped, Halted, Aborted }; +struct RunStep { + RunAction action = RunAction::None; + int addrIdx = -1; +}; +struct RunAddress { + NoteGroup group; + AddressRun run; + AddrOutcome outcome = AddrOutcome::Pending; + bool done = false; + int opsDone = 0; + int transient = 0; + int unlockRefusals = 0; + std::string failure; +}; +struct RunBatch { + int addrIdx = 0; + std::string opid; + int merging = 0; + OpState state = OpState::Unknown; + std::string error, txid; + int misses = 0; + double acceptedAt = 0.0; + double pollableSec = 0.0; // time since accepted in which a status poll could have gone out +}; +struct RunProgress { + int addresses = 0, addressesDone = 0, batchesSubmitted = 0, batchesEstimated = 0, batchesDone = 0, notesMerged = 0; + bool stopping = false; + RunWait wait = RunWait::None; +}; +struct RunSummary { + RunEnd end = RunEnd::None; + int addresses = 0, addressesFailed = 0, addressesUnconfirmed = 0, batchesSubmitted = 0, batchesSucceeded = 0, + notesMerged = 0; + std::string firstError; +}; +RunVerdict runVerdict(const RunSummary& s); +const char* runWaitKey(RunWait wait); // i18n key of the wait line, nullptr when there is nothing to say + +class ConsolidationRun { +public: + bool start(const std::vector& groups, RunConfig cfg, double now); + void requestStop(); + bool abort(); + RunStep next(const RunEnv& env, double now); + void markCallPosted(int addrIdx, double now); + void markPollPosted(double now); + // Returns an orphan opid (an accepted merge whose result is stale) for the caller to track, else "". + std::string onCallResult(std::uint64_t gen, int addrIdx, const MergeCallResult& r, rpc::RpcFailureKind kind, + double callSec, double now); + void onPollResult(std::uint64_t gen, bool ok, const std::vector& reports, double now); + bool consumeLockResyncRequest(); + std::vector outstandingOpids() const; + void dismiss(); + bool active() const { return phase_ == Phase::Running; } + bool hasRun() const { return phase_ != Phase::Idle; } + bool stopping() const { return stopping_; } + std::uint64_t gen() const { return gen_; } + RunWait lastWait() const { return lastWait_; } + RunEnd end() const { return end_; } + RunProgress progress() const; + RunSummary summary() const; + const std::vector& addresses() const { return addrs_; } + const std::vector& batches() const { return batches_; } + const RunConfig& config() const { return cfg_; } + +private: + enum class Phase : std::uint8_t { Idle, Running, Finished }; + static void worsen(RunAddress& a, AddrOutcome o) { if (o > a.outcome) a.outcome = o; } + void endAddress(RunAddress& a, AddrOutcome o, const std::string& msg); + void applyOpState(RunBatch& b, OpState st, const OpReport* rep); + void halt(const std::string& msg) { halted_ = true; if (haltError_.empty()) haltError_ = msg; } + + Phase phase_ = Phase::Idle; + RunEnd end_ = RunEnd::None; + RunConfig cfg_; + std::vector addrs_; + std::vector batches_; + std::uint64_t gen_ = 0; + std::size_t cur_ = 0; + int callAddr_ = -1; + bool callInFlight_ = false, pollInFlight_ = false, stopping_ = false, halted_ = false, lockResync_ = false; + double callPostedAt_ = 0, pollPostedAt_ = 0, nextCallAt_ = 0, nextPollAt_ = 0, syncHoldUntil_ = 0, + unlockHoldUntil_ = 0; + int deadStreak_ = 0, opFailStreak_ = 0; + std::string haltError_; + RunWait lastWait_ = RunWait::None; + double lastPumpAt_ = 0.0; +}; + // ── Scope scan (the Merge/Consolidate dialog's listunspent + z_listunspent) ───────────────────────── // Both calls are whole-wallet scans under the daemon's cs_main. The dialog used to post the pair again on // the frame after any failure, every frame for as long as it stayed open (and forever on Lite, which has diff --git a/src/ui/windows/shield_dialog.cpp b/src/ui/windows/shield_dialog.cpp index e09d7d8..212774b 100644 --- a/src/ui/windows/shield_dialog.cpp +++ b/src/ui/windows/shield_dialog.cpp @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -54,13 +55,9 @@ static bool s_per_address = false; static std::vector s_groups; // spendable shielded notes by address (scope load) static std::set s_selected; // addresses ticked for per-address consolidation static std::set s_unticked; // unticked by the user this opening: kept across a re-scan -struct PerAddressRun { - services::AddressRun run; - std::vector state; // per opid: 0 = running, 1 = done, 2 = failed, 3 = gone from the daemon (unknown) - std::string failure; // first failed op's message -}; -static std::vector s_runs; // after submit -static bool s_runs_submitting = false; +// The per-address run itself is App's (App::consolidation()): it keeps going when the dialog closes. A finished +// one is dismissed when the dialog next opens, so the form starts fresh. +static bool s_dismiss_finished = false; // The op-status polls: one in flight at a time, cleared only by its reply, even across a close and // reopen; a reply for an earlier opening (another session) changes nothing else. static bool s_poll_in_flight = false; @@ -75,8 +72,6 @@ static void resetTransient() s_status_message.clear(); s_operation_id.clear(); s_creating_addr = false; - s_runs.clear(); - s_runs_submitting = false; } // Count + sum spendable transparent UTXOs and shielded notes so the user can see the scope of a @@ -86,10 +81,13 @@ static void loadScope(App* app) { if (!app || !app->worker()) return; const std::uint32_t gen = services::scopeScanBegin(s_scan, services::steadySeconds()); - app->worker()->post([app, rpc = app->rpc(), gen]() -> rpc::RPCWorker::MainCb { + // Two whole-wallet scans the user asked for: never held back, but charged to the scan budget. + const services::ScanTicket ticket = app->beginAdhocWalletScan(); + app->worker()->post([app, rpc = app->rpc(), gen, ticket]() -> rpc::RPCWorker::MainCb { int tC = 0, zC = 0; double tA = 0.0, zA = 0.0; std::string error; rpc::RpcFailureKind kind = rpc::RpcFailureKind::None; std::vector groups; + const auto t0 = std::chrono::steady_clock::now(); try { if (!rpc) throw std::runtime_error("Not connected"); // On a large wallet each scan can outlast the default 30 s; give them 120 s, counted as @@ -113,7 +111,9 @@ static void loadScope(App* app) error = "unknown error"; kind = rpc::RpcFailureKind::Other; } - return [tC, zC, tA, zA, error, kind, gen, groups]() { + const double costMs = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + return [app, ticket, costMs, tC, zC, tA, zA, error, kind, gen, groups]() { + app->endAdhocWalletScan(ticket, costMs, services::chatScanOutcome(error.empty(), kind)); if (!services::scopeScanFinish(s_scan, gen, error, kind, services::steadySeconds())) return; s_t_count = tC; s_z_count = zC; s_t_amount = tA; s_z_amount = zA; s_groups = groups; @@ -141,6 +141,7 @@ void ShieldDialog::show(Mode mode) s_consolidate = false; // reset preset flags so stale statics don't leak across opens s_src = 2; resetTransient(); + s_dismiss_finished = true; s_from_address[0] = '\0'; if (mode == Mode::ShieldCoinbase) strncpy(s_from_address, "*", sizeof(s_from_address)); s_to_address[0] = '\0'; @@ -228,6 +229,7 @@ static void autoSelectDestination(const WalletState& state) // AND kicks the modal's own live-progress poll. static void submitOperation(App* app) { + if (!app->worker()) return; // before the pending state, so a missing worker can't wedge it s_operation_pending = true; s_op_terminal = false; s_status_message = TR("shield_submitting"); @@ -267,12 +269,15 @@ static void submitOperation(App* app) else fromAddrs = { "*" }; std::string to(s_to_address); double fee = s_fee; int limit = s_utxo_limit; - if (!app->worker()) return; - app->worker()->post([app, rpc = app->rpc(), fromAddrs, to, fee, limit]() -> rpc::RPCWorker::MainCb { + // The merge's GetFilteredNotes is a whole-wallet scan: charged to the scan budget, never held back. + const services::ScanTicket ticket = app->beginAdhocWalletScan(); + app->worker()->post([app, rpc = app->rpc(), fromAddrs, to, fee, limit, ticket]() -> rpc::RPCWorker::MainCb { nlohmann::json addrs = nlohmann::json::array(); for (const auto& a : fromAddrs) addrs.push_back(a); nlohmann::json result; std::string error; bool uncertain = false; + rpc::RpcFailureKind kind = rpc::RpcFailureKind::None; + const auto t0 = std::chrono::steady_clock::now(); try { if (!rpc) throw std::runtime_error("Not connected"); // The daemon scans the whole wallet before it returns the opid; on a large wallet that can @@ -281,15 +286,22 @@ static void submitOperation(App* app) App::MaintenanceCallScope maint(*app); rpc::RPCClient::TraceScope trace("Shield dialog / Consolidate"); // fromaddrs, toaddr, fee, transparent_limit, shielded_limit — cap both to the batch size. - result = rpc->call("z_mergetoaddress", {addrs, to, fee, limit, limit}, services::kScopeCallTimeoutSec); + result = rpc->call("z_mergetoaddress", {addrs, to, fee, limit, limit}, services::kRunCallTimeoutSec); } catch (const std::exception& e) { error = e.what(); + if (error.empty()) error = "error"; + kind = rpc::rpcFailureKindOf(e); // Sent, then cut off with no reply: our own abort on a lifecycle disconnect, the reply timeout // or a dropped connection. The daemon keeps running the handler and may still queue the merge. if (const auto* t = dynamic_cast(&e)) uncertain = t->requestSent && t->httpStatus == 0; + } catch (...) { + error = "unknown error"; + kind = rpc::RpcFailureKind::Other; } - return [app, result, error, uncertain]() { + const double ms = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + return [app, result, error, uncertain, ticket, kind, ms]() { + app->endAdhocWalletScan(ticket, ms, services::chatScanOutcome(error.empty(), kind)); if (error.empty()) { s_operation_id = result.value("opid", ""); s_status_message = TR("merge_progress"); @@ -318,134 +330,17 @@ static std::vector selectedGroups() return out; } -// Per-address consolidation. One background job submits, address by address, merges of each address's -// notes back into that address — several batches when it holds more notes than the batch size. -// z_mergetoaddress locks a batch's notes before it returns, so back-to-back calls take disjoint notes. +// Per-address consolidation: App runs it (App::startConsolidation): paced, one merge at a time, and it keeps +// going when this dialog closes. App toasts any refusal. static void submitPerAddress(App* app) { const auto groups = selectedGroups(); - if (groups.empty() || !app->worker()) return; - s_runs_submitting = true; - s_operation_pending = true; - s_op_terminal = false; - s_status_message = TR("shield_submitting"); - s_last_poll = ImGui::GetTime(); - const double fee = s_fee; - const int limit = s_utxo_limit; - app->worker()->post([app, rpc = app->rpc(), groups, fee, limit]() -> rpc::RPCWorker::MainCb { - std::vector runs; - for (const auto& g : groups) { - runs.push_back(services::consolidateInPlace(g.address, [&](const std::string& addr) { - services::MergeCallResult r; - try { - rpc::RPCClient::TraceScope trace("Shield dialog / Consolidate in place"); - // fromaddrs = [addr], toaddr = addr: the address's notes merge back into it. - const nlohmann::json res = rpc->call("z_mergetoaddress", - {nlohmann::json::array({addr}), addr, fee, limit, limit}); - r.opid = res.value("opid", std::string()); - r.merging = res.value("mergingNotes", 0); - r.remaining = res.value("remainingNotes", 0); - if (r.opid.empty()) r.error = "no operation id"; - } catch (const std::exception& e) { r.error = e.what(); } - return r; - })); - } - return [app, runs]() { - s_runs_submitting = false; - s_runs.clear(); - int submitted = 0; - for (const auto& r : runs) { - PerAddressRun pr; - pr.run = r; - pr.state.assign(r.opids.size(), 0); - for (const auto& id : r.opids) { app->trackOperation(id); ++submitted; } - s_runs.push_back(std::move(pr)); - } - if (submitted > 0) { - s_status_message = TR("merge_progress"); - Notifications::instance().success(TR("merge_started")); - } else { - s_operation_pending = false; s_op_terminal = true; - s_status_message = TR("merge_keep_each_nothing_submitted"); - Notifications::instance().error(TR("merge_keep_each_nothing_submitted")); - } - }; - }); -} - -// Progress of a per-address run: poll every ~2s until each batch's operation is done or failed. Filtered -// to the batches still running, as the shared poller does (unfiltered if the daemon rejected the filter). -static void pollPerAddress(App* app) -{ - if (s_runs.empty() || s_op_terminal || s_poll_in_flight || !app->worker()) return; - const double now = ImGui::GetTime(); - if (now - s_last_poll < 2.0) return; - std::vector running; - for (const auto& r : s_runs) - for (size_t i = 0; i < r.run.opids.size(); ++i) - if (r.state[i] == 0) running.push_back(r.run.opids[i]); - const auto filter = services::NetworkRefreshService::buildOperationFilterParams(running); - if (!filter) return; - s_last_poll = now; - s_poll_in_flight = true; - const unsigned session = s_poll_session; - nlohmann::json params = app->opidFilterUnsupported() ? nlohmann::json::array() : *filter; - app->worker()->post([rpc = app->rpc(), params = std::move(params), session]() -> rpc::RPCWorker::MainCb { - nlohmann::json result; std::string error; - try { - if (!rpc) throw std::runtime_error("Not connected"); - rpc::RPCClient::TraceScope trace("Shield dialog / Consolidate in place status"); - result = rpc->call("z_getoperationstatus", params); - } catch (const std::exception& e) { error = e.what(); } - return [result, error, session]() { - s_poll_in_flight = false; - if (session != s_poll_session) return; // a reply for an earlier opening - if (!error.empty() || !result.is_array()) return; // transient — retry next tick - std::set reported; - for (const auto& op : result) { - if (!op.is_object() || !op.contains("id") || !op["id"].is_string()) continue; - const std::string id = op["id"].get(); - reported.insert(id); - const std::string status = (op.contains("status") && op["status"].is_string()) - ? op["status"].get() : std::string(); - for (auto& r : s_runs) { - for (size_t i = 0; i < r.run.opids.size(); ++i) { - if (r.run.opids[i] != id || r.state[i] != 0) continue; - if (status == "success") r.state[i] = 1; - else if (status == "failed" || status == "cancelled" || status == "canceled") { - r.state[i] = 2; - if (r.failure.empty() && op.contains("error") && op["error"].is_object() && - op["error"].contains("message") && op["error"]["message"].is_string()) - r.failure = op["error"]["message"].get(); - } - } - } - } - // A valid reply without one of our running ops: the daemon no longer holds it (restart, or - // popped by z_getoperationresult); its outcome can't be confirmed. One poll at a time, and - // states change only here, so the running set now is the set that was asked for. - for (auto& r : s_runs) - for (size_t i = 0; i < r.run.opids.size(); ++i) - if (r.state[i] == 0 && !reported.count(r.run.opids[i])) r.state[i] = 3; - bool allDone = true; - int failed = 0, unknown = 0; - for (const auto& r : s_runs) - for (int st : r.state) { if (st == 0) allDone = false; if (st == 2) ++failed; if (st == 3) ++unknown; } - if (!allDone) return; - s_operation_pending = false; - s_op_terminal = true; - if (failed > 0) { - s_status_message = TR("merge_keep_each_some_failed"); - Notifications::instance().error(TR("merge_keep_each_some_failed")); - } else if (unknown > 0) { - s_status_message = TR("send_status_unconfirmed"); - Notifications::instance().warning(TR("send_status_unconfirmed")); - } else { - s_status_message = TR("shield_completed"); - Notifications::instance().success(TR("shield_merge_done")); - } - }; - }); + if (groups.empty()) return; + if (app->startConsolidation(groups, s_fee, s_utxo_limit)) { + s_confirm = false; + s_status_message.clear(); // an earlier "Into one" error / uncertain text belongs to that op + s_op_terminal = false; + } } // Live-progress self-poll: while an op is in flight, poll z_getoperationstatus every ~2s so the modal @@ -547,12 +442,16 @@ void ShieldDialog::render(App* app) const auto& state = app->getWalletState(); autoSelectDestination(state); pollOperation(app); - pollPerAddress(app); + const services::ConsolidationRun& run = app->consolidation(); + if (s_dismiss_finished) { // a finished run from an earlier opening: start with a fresh form + s_dismiss_finished = false; + app->dismissConsolidation(); + } if (isMerge) { services::ScopeInputs in; in.mergeMode = true; - in.busyWithOp = !s_operation_id.empty() || s_operation_pending || !s_runs.empty(); - in.nodeReady = app->walletScanReady(); + in.busyWithOp = !s_operation_id.empty() || s_operation_pending || run.hasRun(); + in.nodeReady = app->walletScanReady() && !app->walletScanInFlight(); // don't stack on another wallet scan in.now = services::steadySeconds(); if (services::scopeScanWant(s_scan, in) == services::ScopeWant::Post) loadScope(app); } @@ -563,9 +462,13 @@ void ShieldDialog::render(App* app) : TR("shield_description")); ImGui::Spacing(); - // Submitted — inputs frozen, showing progress. - const bool opInFlight = !s_operation_id.empty() || s_runs_submitting || !s_runs.empty(); + // Submitted — inputs frozen, showing progress. s_operation_pending covers the wait for the opid (up to + // 120 s), when Confirm used to stay armed. + const bool opInFlight = s_operation_pending || !s_operation_id.empty() || (isMerge && run.hasRun()); const bool perAddress = isMerge && s_per_address; + const float rowH = ImGui::GetFrameHeightWithSpacing(); + int maxRows = static_cast(S.drawElement("dialogs.shield", "list-max-rows").size); + if (maxRows < 1) maxRows = 7; // ── Merge: scope + source selector ────────────────────────────────────────────────────────── if (isMerge && !opInFlight) { @@ -620,8 +523,8 @@ void ShieldDialog::render(App* app) if (s_scan.phase == services::ScopePhase::Loaded && listed == 0) { wrappedCaption(material::Warning(), TR("merge_keep_each_none")); } else if (listed > 0) { - const float rowH = ImGui::GetFrameHeightWithSpacing(); - const float listH = std::min(static_cast(listed), 7.0f) * rowH + ImGui::GetStyle().WindowPadding.y * 2.0f; + const float listH = std::min(static_cast(listed), static_cast(maxRows)) * rowH + + ImGui::GetStyle().WindowPadding.y * 2.0f; if (ImGui::BeginChild("##ConsolidateEach", ImVec2(-1, listH), ImGuiChildFlags_Borders)) { for (const auto& g : s_groups) { if (g.notes < 2) continue; @@ -751,51 +654,98 @@ void ShieldDialog::render(App* app) ImGui::Spacing(); } - // ── Per-address progress ──────────────────────────────────────────────────────────────────── - if (!s_runs.empty()) { - bool leftovers = false; - for (const auto& r : s_runs) { - const std::string label = app->getAddressLabel(r.run.address); - const char* name = label.empty() ? nullptr : label.c_str(); - const std::string shortA = shortAddr(r.run.address); - int done = 0, failed = 0, unknown = 0; - for (int st : r.state) { if (st == 1) ++done; if (st == 2) ++failed; if (st == 3) ++unknown; } - const int batches = static_cast(r.run.opids.size()); - char line[320]; - ImU32 col = material::OnSurfaceMedium(); - if (batches == 0) { - std::snprintf(line, sizeof(line), TR("merge_keep_each_status_not_started"), name ? name : shortA.c_str(), - r.run.error.c_str()); - col = material::Error(); - } else if (failed > 0) { - std::snprintf(line, sizeof(line), TR("merge_keep_each_status_failed"), name ? name : shortA.c_str(), - r.failure.c_str()); - col = material::Error(); - } else if (unknown > 0 && done + unknown == batches) { - std::snprintf(line, sizeof(line), TR("merge_keep_each_status_unconfirmed"), name ? name : shortA.c_str()); - col = material::Warning(); - } else if (done == batches) { - std::snprintf(line, sizeof(line), TR("merge_keep_each_status_done"), name ? name : shortA.c_str(), - r.run.notesMerged, batches); - col = material::Success(); - } else { - std::snprintf(line, sizeof(line), TR("merge_keep_each_status_running"), name ? name : shortA.c_str(), - r.run.notesMerged, done, batches); + // ── Per-address run (App-owned) ──────────────────────────────────────────────────────────── + if (isMerge && run.hasRun()) { + const services::RunProgress p = run.progress(); + const services::RunSummary sum = run.summary(); + const bool live = run.active(); + if (live) { + char cap[400]; + if (p.stopping) std::snprintf(cap, sizeof(cap), "%s", TR("merge_run_stopping")); + else std::snprintf(cap, sizeof(cap), TR("merge_run_progress_fmt"), p.batchesSubmitted, + std::max(1, p.batchesEstimated), p.notesMerged); + wrappedCaption(material::OnSurfaceMedium(), cap); + if (!p.stopping) { + if (const char* key = services::runWaitKey(run.lastWait())) wrappedCaption(material::OnSurfaceMedium(), TR(key)); } - wrappedCaption(col, line); - if (batches > 1 || r.run.remaining >= 2) leftovers = true; } + ImGui::ProgressBar(static_cast(p.batchesDone) / static_cast(std::max(1, p.batchesEstimated)), + ImVec2(-1, 0)); + const auto& addrs = run.addresses(); + const float runListH = std::min(static_cast(addrs.size()), static_cast(maxRows)) * rowH + + ImGui::GetStyle().WindowPadding.y * 2.0f; + bool leftovers = false; + if (ImGui::BeginChild("##ConsolidateRuns", ImVec2(-1, runListH), ImGuiChildFlags_Borders)) { + for (const auto& a : addrs) { + const std::string label = app->getAddressLabel(a.group.address); + const std::string name = label.empty() ? shortAddr(a.group.address) : label; + const int nOps = static_cast(a.run.opids.size()); + char line[400]; + ImU32 col = material::OnSurfaceMedium(); + if (nOps == 0 && !a.done) { + // Stopping: an address never called will not be sent; one with a call in flight still may be. + const bool willRun = live && !(p.stopping && a.outcome == services::AddrOutcome::Pending); + std::snprintf(line, sizeof(line), willRun ? TR("merge_run_status_waiting") : TR("merge_run_status_not_run"), + name.c_str(), a.group.notes); + } else if (a.outcome == services::AddrOutcome::Uncertain) { // a call or batch whose result is unknown + std::snprintf(line, sizeof(line), TR("merge_keep_each_status_unconfirmed"), name.c_str()); + col = material::Warning(); + } else if (nOps == 0) { + std::snprintf(line, sizeof(line), TR("merge_keep_each_status_not_started"), name.c_str(), + a.run.error.empty() ? TR("shield_unknown_error") : a.run.error.c_str()); + col = material::Error(); + } else if (a.outcome == services::AddrOutcome::Failed) { + const std::string why = !a.failure.empty() ? a.failure + : !a.run.error.empty() ? a.run.error : std::string(TR("shield_unknown_error")); + std::snprintf(line, sizeof(line), TR("merge_keep_each_status_failed"), name.c_str(), why.c_str()); + col = material::Error(); + } else if ((a.done || !live) && a.opsDone == nOps) { // an ended run: every batch is terminal + std::snprintf(line, sizeof(line), TR("merge_keep_each_status_done"), name.c_str(), a.run.notesMerged, nOps); + col = a.done ? material::Success() : material::OnSurfaceMedium(); // stopped part-way: not complete + } else { // only while live: batches outstanding or more to submit + std::snprintf(line, sizeof(line), TR("merge_keep_each_status_running"), name.c_str(), a.run.notesMerged, + a.opsDone, nOps); + } + wrappedCaption(col, line); + if (nOps > 1 || a.run.remaining >= 2) leftovers = true; + } + } + ImGui::EndChild(); if (leftovers) { ImGui::Spacing(); - ImGui::PushStyleColor(ImGuiCol_Text, ImGui::ColorConvertU32ToFloat4(material::OnSurfaceMedium())); - ImGui::TextWrapped("%s", TR("merge_keep_each_run_again")); - ImGui::PopStyleColor(); + wrappedCaption(material::OnSurfaceMedium(), TR("merge_keep_each_run_again")); + } + if (!live) { // below the rows: merge_keep_each_some_failed says "see the list above" + std::string endLine; + ImU32 endCol = material::OnSurfaceMedium(); + switch (services::runVerdict(sum)) { + case services::RunVerdict::Clean: endLine = TR("shield_completed"); endCol = material::Success(); break; + case services::RunVerdict::NothingSubmitted: endLine = TR("merge_keep_each_nothing_submitted"); endCol = material::Error(); break; + case services::RunVerdict::SomeFailed: endLine = TR("merge_keep_each_some_failed"); endCol = material::Error(); break; + case services::RunVerdict::Unconfirmed: endLine = TR("send_status_unconfirmed"); endCol = material::Warning(); break; + case services::RunVerdict::Stopped: endLine = TR("merge_run_stopped_toast"); break; + case services::RunVerdict::Halted: { + std::string err = sum.firstError.empty() ? std::string(TR("shield_unknown_error")) : sum.firstError; + if (err.size() > 160) err = err.substr(0, 160) + "..."; + char hb[512]; + std::snprintf(hb, sizeof(hb), TR("merge_run_halted_fmt"), err.c_str()); + endLine = hb; + endCol = material::Error(); + break; + } + case services::RunVerdict::Aborted: endLine = TR("merge_run_aborted_toast"); endCol = material::Warning(); break; + default: break; + } + if (!endLine.empty()) { + ImGui::Spacing(); + wrappedCaption(endCol, endLine.c_str()); + } } ImGui::Spacing(); } // ── Live progress / status ────────────────────────────────────────────────────────────────── - if (!s_status_message.empty()) { + if (!s_status_message.empty() && !(isMerge && run.hasRun())) { // the single-op path's message, not the run's if (s_operation_pending && !s_op_terminal) ImGui::TextColored(ImVec4(1.0f, 0.8f, 0.0f, 1.0f), "%s", s_status_message.c_str()); else @@ -808,11 +758,34 @@ void ShieldDialog::render(App* app) const bool syncing = state.sync.syncing; const bool haveDest = s_to_address[0] != '\0'; + if (isMerge && run.hasRun()) { + // The run's footer: Stop + Close while it runs (it keeps going in the background), Done once it ended. + ImFont* bf = S.resolveFont(cancelBtn.font); + if (run.active()) { + material::BeginOverlayDialogFooter((shieldBtn.width + cancelBtn.width) * dp + ImGui::GetStyle().ItemSpacing.x, + /*drawSeparator=*/false); + ImGui::BeginDisabled(run.stopping()); + if (material::TactileButton(TR("merge_run_stop"), ImVec2(shieldBtn.width * dp, 0), S.resolveFont(shieldBtn.font))) + app->stopConsolidation(); + ImGui::EndDisabled(); + ImGui::SameLine(); + if (material::TactileButton(TR("close"), ImVec2(cancelBtn.width * dp, 0), bf)) s_open = false; + } else { + material::BeginOverlayDialogFooter(cancelBtn.width * dp, /*drawSeparator=*/false); + if (material::TactileButton(TR("done"), ImVec2(cancelBtn.width * dp, 0), bf)) { + app->dismissConsolidation(); + s_open = false; + } + } + material::EndOverlayDialog(); + return; + } + if (opInFlight) { // After submit: just a Close button (progress shows above; op continues in the background). - material::BeginOverlayDialogFooter(cancelBtn.width, /*drawSeparator=*/false); + material::BeginOverlayDialogFooter(cancelBtn.width * dp, /*drawSeparator=*/false); if (material::TactileButton(s_op_terminal ? TR("done") : TR("close"), - ImVec2(cancelBtn.width, 0), S.resolveFont(cancelBtn.font))) + ImVec2(cancelBtn.width * dp, 0), S.resolveFont(cancelBtn.font))) s_open = false; material::EndOverlayDialog(); return; @@ -848,11 +821,11 @@ void ShieldDialog::render(App* app) if (isMerge && scopeLoaded && srcCount() == 0) can_submit = false; if (perAddress) can_submit = connected && !syncing && scopeLoaded && !selectedGroups().empty(); - float footerBtnW = shieldBtn.width + cancelBtn.width + ImGui::GetStyle().ItemSpacing.x; + float footerBtnW = (shieldBtn.width + cancelBtn.width) * dp + ImGui::GetStyle().ItemSpacing.x; material::BeginOverlayDialogFooter(footerBtnW, /*drawSeparator=*/false); if (!can_submit) ImGui::BeginDisabled(); - if (material::TactileButton(primaryLabel, ImVec2(shieldBtn.width, 0), S.resolveFont(shieldBtn.font))) { + if (material::TactileButton(primaryLabel, ImVec2(shieldBtn.width * dp, 0), S.resolveFont(shieldBtn.font))) { if (s_confirm) { if (perAddress) submitPerAddress(app); else submitOperation(app); } else { s_confirm = true; } // first click → show the confirm summary } @@ -860,13 +833,25 @@ void ShieldDialog::render(App* app) if (!can_submit && ImGui::IsItemHovered(ImGuiHoveredFlags_AllowWhenDisabled)) { if (!connected) material::Tooltip("%s", TR("send_tooltip_not_connected")); else if (syncing) material::Tooltip("%s", TR("send_tooltip_syncing")); + else if (perAddress && s_scan.phase != services::ScopePhase::Loaded) { + // Why the list isn't there yet, not "select addresses". + char why[320]; + if (s_scan.phase == services::ScopePhase::Failed) { + const std::string reason = s_scan.error.size() > 160 ? s_scan.error.substr(0, 160) + "..." : s_scan.error; + std::snprintf(why, sizeof(why), TR("merge_scope_failed_fmt"), reason.c_str()); + } else { + std::snprintf(why, sizeof(why), "%s", s_scan.phase == services::ScopePhase::Loading ? TR("merge_scope_loading") + : state.isLocked() ? TR("merge_wait_locked") : TR("merge_waiting_node")); + } + material::Tooltip("%s", why); + } else if (perAddress) material::Tooltip("%s", TR("merge_keep_each_select")); else if (!haveDest) material::Tooltip("%s", TR("shield_select_z")); else if (isMerge && srcCount() == 0) material::Tooltip("%s", TR("merge_no_spendable")); } ImGui::SameLine(); - if (material::TactileButton(secondaryLabel, ImVec2(cancelBtn.width, 0), S.resolveFont(cancelBtn.font))) { + if (material::TactileButton(secondaryLabel, ImVec2(cancelBtn.width * dp, 0), S.resolveFont(cancelBtn.font))) { if (s_confirm) s_confirm = false; // Back → return to the form else s_open = false; // Cancel → close } diff --git a/src/util/i18n.cpp b/src/util/i18n.cpp index 5de2ba3..6d3767e 100644 --- a/src/util/i18n.cpp +++ b/src/util/i18n.cpp @@ -2543,6 +2543,22 @@ void I18n::loadBuiltinEnglish() strings_["merge_keep_each_status_running"] = "%s — merging %d notes (%d of %d batches done)…"; strings_["merge_keep_each_status_unconfirmed"] = "%s \xE2\x80\x94 status could not be confirmed"; strings_["merge_keep_each_run_again"] = "Addresses merged in several batches now hold one note per batch. Run this again after those confirm to merge them into one."; + strings_["merge_run_progress_fmt"] = "Consolidating\xE2\x80\xA6 batch %d of about %d, %d notes merged so far. You can close this window; it keeps running in the background."; + strings_["merge_run_stop"] = "Stop"; + strings_["merge_run_stopping"] = "Stopping after the batches already sent\xE2\x80\xA6"; + strings_["merge_run_status_waiting"] = "%s \xE2\x80\x94 waiting (%d notes)"; + strings_["merge_run_status_not_run"] = "%s \xE2\x80\x94 not consolidated (%d notes)"; + strings_["merge_run_wait_sync"] = "Paused: the node is still syncing."; + strings_["merge_run_wait_send"] = "Paused until your pending transaction goes through."; + strings_["merge_run_wait_batch"] = "Spacing the batches so the node stays responsive\xE2\x80\xA6"; + strings_["merge_run_err_timeout"] = "The node did not answer in time; the batch may still be queued."; + strings_["merge_run_stopped_toast"] = "Consolidation stopped. Run it again to merge the rest."; + strings_["merge_run_aborted_toast"] = "The node connection was reset, so consolidation stopped. Run it again to continue."; + strings_["merge_run_start_failed"] = "Couldn't start the consolidation \xE2\x80\x94 the node isn't connected."; + strings_["merge_run_chip_fmt"] = "Consolidating %d/%d"; + strings_["merge_run_halted_fmt"] = "Consolidation stopped after repeated failures: %s"; + strings_["merge_run_busy"] = "A consolidation is running. Wait for it to finish, or stop it, first."; + strings_["merge_run_migration_pending"] = "Finish or cancel the seed migration before consolidating."; strings_["done"] = "Done"; // --- Transaction Details Dialog --- diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index 78902c2..5676779 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -4521,6 +4521,10 @@ void testConsolidationStringsTranslated() "merge_keep_each_status_done", "merge_keep_each_status_running", "merge_keep_each_run_again", "merge_waiting_node", "merge_wait_locked", "merge_scope_failed_fmt", "retry", "merge_keep_each_status_unconfirmed", "merge_submit_uncertain", + "merge_run_progress_fmt", "merge_run_stop", "merge_run_stopping", "merge_run_status_waiting", + "merge_run_status_not_run", "merge_run_wait_sync", "merge_run_wait_send", "merge_run_wait_batch", + "merge_run_err_timeout", "merge_run_stopped_toast", "merge_run_aborted_toast", "merge_run_start_failed", + "merge_run_chip_fmt", "merge_run_halted_fmt", "merge_run_busy", "merge_run_migration_pending", }); } @@ -11613,6 +11617,613 @@ void testChatScanGovernorStreams() EXPECT_TRUE(static_cast(gov.tryAdmit(ScanStream::Balance, true, a, 10.0))); } +// C210: reading z_mergetoaddress's reply, and one call into an address's run. +void testMergeCallResult() +{ + using namespace dragonx::services; + using nlohmann::json; + MergeCallResult r; + readMergeCallResult(json{{"opid", "o"}, {"mergingNotes", 50}, {"remainingNotes", 70}}, r); + EXPECT_TRUE(r.opid == "o" && r.merging == 50 && r.remaining == 70 && r.error.empty()); + readMergeCallResult(json{{"opid", "o"}, {"mergingNotes", "50"}}, r); + EXPECT_TRUE(r.opid == "o" && r.merging == 0 && r.error.empty()); // accepted, count unreadable + for (const json& bad : {json{{"mergingNotes", 3}}, json{{"opid", 5}}, json{{"opid", ""}}, json::array(), json("x"), json()}) { + readMergeCallResult(bad, r); + EXPECT_TRUE(r.error == "no operation id" && r.opid.empty()); + } + readMergeCallResult(json{{"opid", "o"}, {"remainingNotes", -4}}, r); + EXPECT_EQ(r.remaining, 0); + readMergeCallResult(json{{"opid", "o"}, {"remainingNotes", static_cast(1) << 40}}, r); + EXPECT_EQ(r.remaining, INT_MAX); + + AddressRun run; + MergeCallResult ok{"o1", 50, 70, ""}; + EXPECT_TRUE(recordMergeResult(run, ok, 20)); + EXPECT_FALSE(recordMergeResult(run, MergeCallResult{"o2", 50, 1, ""}, 20)); // one left: done + AddressRun capped; + EXPECT_FALSE(recordMergeResult(capped, ok, 1)); // at maxBatches + AddressRun failed; + EXPECT_FALSE(recordMergeResult(failed, MergeCallResult{"", 0, 0, "boom"}, 20)); + EXPECT_EQ(failed.error, std::string("boom")); + int calls = 0; + consolidateInPlace("z", [&](const std::string&) { ++calls; return ok; }, 0); + EXPECT_EQ(calls, 0); +} + +void testOpReports() +{ + using namespace dragonx::services; + using nlohmann::json; + EXPECT_TRUE(parseOpReports(json::object()).empty()); + const json reply = json::array({ + json(5), + json{{"status", "success"}}, + json{{"id", 7}, {"status", "success"}}, + json{{"id", ""}, {"status", "success"}}, + json{{"id", "a"}, {"status", 3}, {"error", "x"}, {"result", "y"}}, + json{{"id", "b"}, {"status", "failed"}, {"error", json{{"message", "bad"}}}}, + json{{"id", "c"}, {"status", "success"}, {"result", json{{"txid", "tx-c"}}}}, + }); + auto all = parseOpReports(reply); + EXPECT_EQ(all.size(), static_cast(3)); + EXPECT_TRUE(all[0].id == "a" && all[0].status.empty() && all[0].error.empty() && all[0].txid.empty()); + EXPECT_EQ(all[1].error, std::string("bad")); + EXPECT_EQ(all[2].txid, std::string("tx-c")); + auto wanted = parseOpReports(reply, {"c"}); + EXPECT_TRUE(wanted.size() == 1 && wanted[0].id == "c"); + EXPECT_TRUE(opStateFromStatus("queued") == OpState::Queued); + EXPECT_TRUE(opStateFromStatus("executing") == OpState::Executing); + EXPECT_TRUE(opStateFromStatus("success") == OpState::Success); + for (const char* f : {"failed", "cancelled", "canceled"}) EXPECT_TRUE(opStateFromStatus(f) == OpState::Failed); + EXPECT_TRUE(opStateFromStatus("") == OpState::Unknown); + EXPECT_TRUE(opStateFromStatus("x") == OpState::Unknown); +} + +void testMergeFailureClassification() +{ + using namespace dragonx::services; + using K = dragonx::rpc::RpcFailureKind; + EXPECT_TRUE(classifyMergeFailure(K::None, "") == MergeFailure::Uncertain); + for (K k : {K::Deferred, K::NotListening, K::ConnectTimeout, K::Warmup, K::QueueFull}) + EXPECT_TRUE(classifyMergeFailure(k, "x") == MergeFailure::NotSent); + EXPECT_TRUE(classifyMergeFailure(K::ReplyTimeout, "x") == MergeFailure::Uncertain); + EXPECT_TRUE(classifyMergeFailure(K::Aborted, "x") == MergeFailure::Uncertain); + EXPECT_TRUE(classifyMergeFailure(K::ShuttingDown, "x") == MergeFailure::Transient); + EXPECT_TRUE(classifyMergeFailure(K::Unhandled, "x") == MergeFailure::Transient); + EXPECT_TRUE(classifyMergeFailure(K::Auth, "x") == MergeFailure::Fatal); + EXPECT_TRUE(classifyMergeFailure(K::Other, "Not connected") == MergeFailure::NotSent); + // By kind and message alone, an unclassified error is fatal... + EXPECT_TRUE(classifyMergeFailure(K::Other, "RPC request failed: Failure when receiving data from the peer") == MergeFailure::Fatal); + { // ...but a call that was sent and then cut off before any HTTP status may have queued the merge. + using dragonx::rpc::RpcTransportError; + const RpcTransportError cut(K::Other, "RPC request failed: Failure when receiving data from the peer", CURLE_RECV_ERROR, 0, true); + EXPECT_TRUE(mergeCallFailureKind(cut) == K::ReplyTimeout); + EXPECT_TRUE(classifyMergeFailure(mergeCallFailureKind(cut), cut.what()) == MergeFailure::Uncertain); + const RpcTransportError unsent(K::Other, "RPC request failed: x", CURLE_RECV_ERROR, 0, false); + EXPECT_TRUE(mergeCallFailureKind(unsent) == K::Other); + const RpcTransportError http(K::Other, "RPC error: HTTP 504", CURLE_OK, 504, true); + EXPECT_TRUE(mergeCallFailureKind(http) == K::Other); + EXPECT_TRUE(mergeCallFailureKind(std::runtime_error("Not connected")) == K::Other); + EXPECT_TRUE(classifyMergeFailure(mergeCallFailureKind(std::runtime_error("Not connected")), "Not connected") == MergeFailure::NotSent); + const RpcTransportError aborted(K::Aborted, "RPC aborted: disconnecting", CURLE_OK, 0, false); + EXPECT_TRUE(mergeCallFailureKind(aborted) == K::Aborted); + } + const std::pair cases[] = { + {"z_mergetoaddress: Extreme Privacy! Chain still syncing at height 5, aborting to prevent linkability analysis. Please wait until FULLY SYNCED and try again.", MergeFailure::Syncing}, + {"Could not find spending key for payment address.", MergeFailure::NeedsUnlock}, + {"Error: Please enter the wallet passphrase with walletpassphrase first.", MergeFailure::NeedsUnlock}, + {"Could not find any funds to merge.", MergeFailure::NothingLeft}, + {"Destination address is also the only source address, and all its funds are already merged.", MergeFailure::NothingLeft}, + {"Insufficient funds, have 0.00005, which is less than miners fee 0.0001", MergeFailure::NothingLeft}, + {"Fee 0.0001 is greater than the net amount to be shielded 0.00005", MergeFailure::NothingLeft}, + {"Insufficient funds to pay the fee", MergeFailure::NothingLeft}, + {"Async RPC queue is shutting down; the operation was not queued", MergeFailure::Transient}, + {"Invalid parameter, unknown address format", MergeFailure::Fatal}, + }; + for (const auto& [msg, want] : cases) EXPECT_TRUE(classifyMergeFailure(K::AppError, msg) == want); + + EXPECT_TRUE(runWaitKey(RunWait::None) == nullptr && runWaitKey(RunWait::InFlight) == nullptr); + EXPECT_EQ(std::string(runWaitKey(RunWait::Node)), std::string("merge_waiting_node")); + EXPECT_EQ(std::string(runWaitKey(RunWait::Busy)), std::string("merge_waiting_node")); + EXPECT_EQ(std::string(runWaitKey(RunWait::Syncing)), std::string("merge_run_wait_sync")); + EXPECT_EQ(std::string(runWaitKey(RunWait::Locked)), std::string("merge_wait_locked")); + EXPECT_EQ(std::string(runWaitKey(RunWait::UserSend)), std::string("merge_run_wait_send")); + EXPECT_EQ(std::string(runWaitKey(RunWait::Scan)), std::string("merge_run_wait_batch")); + EXPECT_EQ(std::string(runWaitKey(RunWait::Batch)), std::string("merge_run_wait_batch")); + RunSummary sum; + EXPECT_TRUE(runVerdict(sum) == RunVerdict::None); + sum.end = RunEnd::Stopped; EXPECT_TRUE(runVerdict(sum) == RunVerdict::Stopped); + sum.end = RunEnd::Halted; EXPECT_TRUE(runVerdict(sum) == RunVerdict::Halted); + sum.end = RunEnd::Aborted; EXPECT_TRUE(runVerdict(sum) == RunVerdict::Aborted); + sum.end = RunEnd::Completed; + EXPECT_TRUE(runVerdict(sum) == RunVerdict::NothingSubmitted); + sum.batchesSubmitted = 3; + EXPECT_TRUE(runVerdict(sum) == RunVerdict::Clean); + sum.addressesUnconfirmed = 1; + EXPECT_TRUE(runVerdict(sum) == RunVerdict::Unconfirmed); + sum.addressesFailed = 1; + EXPECT_TRUE(runVerdict(sum) == RunVerdict::SomeFailed); +} + +namespace { +dragonx::services::RunStep stepRun(dragonx::services::ConsolidationRun& run, const dragonx::services::RunEnv& env, double now) +{ + using namespace dragonx::services; + const RunStep st = run.next(env, now); + if (st.action == RunAction::Call) run.markCallPosted(st.addrIdx, now); + else if (st.action == RunAction::Poll) run.markPollPosted(now); + return st; +} +dragonx::services::OpReport opRep(const std::string& id, const std::string& status, const std::string& error = "", + const std::string& txid = "") +{ + dragonx::services::OpReport r; + r.id = id; + r.status = status; + r.error = error; + r.txid = txid; + return r; +} +dragonx::services::RunEnv okRunEnv() +{ + dragonx::services::RunEnv env; + env.nodeUsable = true; + return env; +} +} // namespace + +void testConsolidationRun() +{ + using namespace dragonx::services; + using K = dragonx::rpc::RpcFailureKind; + const RunEnv ok = okRunEnv(); + { // Start and dismiss. + ConsolidationRun run; + EXPECT_FALSE(run.start({}, RunConfig{}, 0.0)); + EXPECT_FALSE(run.start({{"z1", 1, 5.0}}, RunConfig{}, 0.0)); // one note + EXPECT_FALSE(run.start({{"z1", 5, 0.0001}}, RunConfig{}, 0.0)); // too small for the fee + RunConfig small; + small.limit = 1; + EXPECT_TRUE(run.start({{"z1", 5, 5.0}}, small, 0.0)); + EXPECT_EQ(run.config().limit, 2); + EXPECT_FALSE(run.start({{"z2", 5, 5.0}}, RunConfig{}, 0.0)); // already running + run.dismiss(); + EXPECT_TRUE(run.active()); // no-op while running + const auto gen1 = run.gen(); + run.requestStop(); + EXPECT_TRUE(stepRun(run, okRunEnv(), 1.0).action == RunAction::Finish); + EXPECT_TRUE(run.end() == RunEnd::Stopped); + run.dismiss(); + EXPECT_FALSE(run.hasRun()); + EXPECT_EQ(run.gen(), gen1); // kept: stale results never match later + EXPECT_TRUE(run.start({{"z1", 5, 5.0}}, RunConfig{}, 2.0)); + EXPECT_TRUE(run.gen() > gen1); + } + { // A normal run: call, poll, call after the gap and the op gate, ..., one Finish. + ConsolidationRun run; + EXPECT_TRUE(run.start({{"zA", 120, 60.0}}, RunConfig{}, 0.0)); + auto st = stepRun(run, ok, 0.0); + EXPECT_TRUE(st.action == RunAction::Call && st.addrIdx == 0); + EXPECT_TRUE(stepRun(run, ok, 0.5).action == RunAction::None); // one RPC at a time + EXPECT_TRUE(run.lastWait() == RunWait::InFlight); + run.onCallResult(run.gen(), 0, MergeCallResult{"o1", 50, 70, ""}, K::None, 1.0, 1.0); + EXPECT_TRUE(stepRun(run, ok, 1.0).action == RunAction::Poll); // learn its state first + run.onPollResult(run.gen(), true, {opRep("o1", "queued")}, 1.1); + EXPECT_TRUE(stepRun(run, ok, 3.5).action == RunAction::None); // past the gap, but still queued + EXPECT_TRUE(run.lastWait() == RunWait::Batch); + EXPECT_TRUE(stepRun(run, ok, 4.1).action == RunAction::Poll); + run.onPollResult(run.gen(), true, {opRep("o1", "executing")}, 4.2); + st = stepRun(run, ok, 4.3); + EXPECT_TRUE(st.action == RunAction::Call); + run.onCallResult(run.gen(), 0, MergeCallResult{"o2", 50, 20, ""}, K::None, 1.0, 5.0); + EXPECT_TRUE(stepRun(run, ok, 5.0).action == RunAction::Poll); + run.onPollResult(run.gen(), true, {opRep("o1", "success", "", "tx1"), opRep("o2", "executing")}, 5.1); + EXPECT_TRUE(stepRun(run, ok, 7.0).action == RunAction::Call); + run.onCallResult(run.gen(), 0, MergeCallResult{"o3", 20, 0, ""}, K::None, 1.0, 8.0); + EXPECT_TRUE(run.addresses()[0].done); + double t = 8.0; + int finishes = 0; + for (; t < 60.0; t += 0.5) { + const auto s2 = stepRun(run, ok, t); + if (s2.action == RunAction::Poll) + run.onPollResult(run.gen(), true, {opRep("o2", "success"), opRep("o3", "success")}, t + 0.1); + if (s2.action == RunAction::Finish) ++finishes; + EXPECT_TRUE(s2.action != RunAction::Call); + } + EXPECT_EQ(finishes, 1); + EXPECT_TRUE(run.end() == RunEnd::Completed); + const auto sum = run.summary(); + EXPECT_EQ(sum.batchesSucceeded, 3); + EXPECT_EQ(sum.notesMerged, 120); + EXPECT_TRUE(runVerdict(sum) == RunVerdict::Clean); + EXPECT_TRUE(run.next(ok, 70.0).action == RunAction::None); + } + { // The gap is the call's own cost, within [2 s, 120 s]. + for (const auto& [callSec, gap] : {std::pair{0.1, 2.0}, {15.0, 15.0}, {200.0, 120.0}}) { + ConsolidationRun run; + run.start({{"zA", 200, 100.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + run.onCallResult(run.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, callSec, 1.0); + // Advance to `t`, answering polls (the op stays Executing); the first Call's time. + double firstCall = -1.0; + for (double t = 1.0; t <= 1.0 + gap + 1.0 && firstCall < 0.0; t += 0.01) { + RunStep s2 = stepRun(run, ok, t); + while (s2.action == RunAction::Poll) { // a due poll goes first + run.onPollResult(run.gen(), true, {opRep("o1", "executing")}, t); + s2 = stepRun(run, ok, t); + } + if (s2.action == RunAction::Call) firstCall = t; + } + EXPECT_NEAR(firstCall, 1.0 + gap, 0.02); + } + } + { // NotSent never ends the address; each retry is at least 5 s later. + ConsolidationRun run; + run.start({{"zA", 10, 5.0}}, RunConfig{}, 0.0); + double t = 0.0, lastCall = -100.0; + int calls = 0; + while (calls < 50 && t < 1000.0) { + const auto s2 = stepRun(run, ok, t); + if (s2.action == RunAction::Call) { + EXPECT_TRUE(t - lastCall >= 5.0); + lastCall = t; + ++calls; + run.onCallResult(run.gen(), 0, MergeCallResult{"", 0, 0, "held back"}, K::Deferred, 0.01, t); + } + t += 0.5; + } + EXPECT_EQ(calls, 50); + EXPECT_FALSE(run.addresses()[0].done); + } + { // Syncing holds 30 s; NeedsUnlock holds 30 s and asks for one lock re-read, then fails the address. + ConsolidationRun run; + run.start({{"zA", 10, 5.0}, {"zB", 10, 5.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + run.onCallResult(run.gen(), 0, MergeCallResult{"", 0, 0, "Chain still syncing at height 5"}, K::AppError, 0.1, 0.1); + EXPECT_TRUE(stepRun(run, ok, 29.0).action == RunAction::None); + EXPECT_TRUE(run.lastWait() == RunWait::Syncing); + EXPECT_TRUE(run.addresses()[0].transient == 0 && run.addresses()[0].unlockRefusals == 0); + EXPECT_TRUE(stepRun(run, ok, 30.2).action == RunAction::Call); + run.onCallResult(run.gen(), 0, MergeCallResult{"", 0, 0, "Could not find spending key for payment address."}, + K::AppError, 0.1, 30.3); + EXPECT_TRUE(run.consumeLockResyncRequest()); + EXPECT_FALSE(run.consumeLockResyncRequest()); + EXPECT_TRUE(stepRun(run, ok, 59.0).action == RunAction::None); + EXPECT_TRUE(run.lastWait() == RunWait::Locked); + EXPECT_TRUE(stepRun(run, ok, 60.5).action == RunAction::Call); + run.onCallResult(run.gen(), 0, MergeCallResult{"", 0, 0, "Could not find spending key for payment address."}, + K::AppError, 0.1, 60.6); + EXPECT_TRUE(run.addresses()[0].done && run.addresses()[0].outcome == AddrOutcome::Failed); + const auto s3 = stepRun(run, ok, 63.0); + EXPECT_TRUE(s3.action == RunAction::Call && s3.addrIdx == 1); + } + { // NothingLeft: Done after a batch, Failed without one. Uncertain is never resent. + ConsolidationRun run; + run.start({{"zA", 10, 5.0}, {"zB", 10, 5.0}, {"zC", 10, 5.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + run.onCallResult(run.gen(), 0, MergeCallResult{"o1", 5, 5, ""}, K::None, 0.1, 0.1); + stepRun(run, ok, 0.2); + run.onPollResult(run.gen(), true, {opRep("o1", "executing")}, 0.2); + EXPECT_TRUE(stepRun(run, ok, 2.5).action == RunAction::Call); + run.onCallResult(run.gen(), 0, MergeCallResult{"", 0, 0, "Could not find any funds to merge."}, K::AppError, 0.1, 2.6); + EXPECT_TRUE(run.addresses()[0].outcome == AddrOutcome::Done); + double t = 2.6; + RunStep s2; + do { t += 0.5; s2 = stepRun(run, ok, t); if (s2.action == RunAction::Poll) run.onPollResult(run.gen(), true, {opRep("o1", "executing")}, t); } + while (s2.action != RunAction::Call && t < 100.0); + EXPECT_EQ(s2.addrIdx, 1); + run.onCallResult(run.gen(), 1, MergeCallResult{"", 0, 0, "Could not find any funds to merge."}, K::AppError, 0.1, t); + EXPECT_TRUE(run.addresses()[1].outcome == AddrOutcome::Failed); + do { t += 0.5; s2 = stepRun(run, ok, t); if (s2.action == RunAction::Poll) run.onPollResult(run.gen(), true, {opRep("o1", "executing")}, t); } + while (s2.action != RunAction::Call && t < 200.0); + EXPECT_EQ(s2.addrIdx, 2); + run.onCallResult(run.gen(), 2, MergeCallResult{"", 0, 0, "RPC error: reply timeout"}, K::ReplyTimeout, 120.0, t); + EXPECT_TRUE(run.addresses()[2].done && run.addresses()[2].outcome == AddrOutcome::Uncertain); + for (int i = 0; i < 400; ++i) { + t += 0.5; + s2 = stepRun(run, ok, t); + EXPECT_TRUE(s2.action != RunAction::Call); + if (s2.action == RunAction::Poll) run.onPollResult(run.gen(), true, {opRep("o1", "success")}, t); + } + EXPECT_TRUE(run.end() == RunEnd::Completed); + const auto sum = run.summary(); + EXPECT_EQ(sum.addressesUnconfirmed, 1); + EXPECT_TRUE(runVerdict(sum) == RunVerdict::SomeFailed); // zB failed (worse than unconfirmed) + } + { // Breakers: three dead addresses in a row halt the run; an accepted call between resets the streak. + ConsolidationRun run; + run.start({{"z1", 10, 5.0}, {"z2", 10, 5.0}, {"z3", 10, 5.0}, {"z4", 10, 5.0}}, RunConfig{}, 0.0); + double t = 0.0; + for (int i = 0; i < 3; ++i) { + RunStep s2; + do { s2 = stepRun(run, ok, t); t += 0.5; } while (s2.action != RunAction::Call); + run.onCallResult(run.gen(), s2.addrIdx, MergeCallResult{"", 0, 0, "Invalid parameter, unknown address format"}, + K::AppError, 0.1, t); + } + RunStep fin; + do { fin = stepRun(run, ok, t); t += 0.5; } while (fin.action == RunAction::None && t < 100.0); + EXPECT_TRUE(fin.action == RunAction::Finish && run.end() == RunEnd::Halted); + EXPECT_FALSE(run.summary().firstError.empty()); + EXPECT_FALSE(run.addresses()[3].done); // never tried + } + { // Op-level breaker: three failed ops in a row. + ConsolidationRun run; + run.start({{"z1", 10, 5.0}, {"z2", 10, 5.0}, {"z3", 10, 5.0}, {"z4", 10, 5.0}}, RunConfig{}, 0.0); + double t = 0.0; + for (int i = 0; i < 3; ++i) { + RunStep s2; + do { + s2 = stepRun(run, ok, t); + if (s2.action == RunAction::Poll) { + std::vector reps; + for (const auto& id : run.outstandingOpids()) reps.push_back(opRep(id, "failed", "proof failed")); + run.onPollResult(run.gen(), true, reps, t); + } + t += 0.5; + } while (s2.action != RunAction::Call && t < 500.0); + if (s2.action == RunAction::Call) + run.onCallResult(run.gen(), s2.addrIdx, MergeCallResult{"op" + std::to_string(i), 9, 1, ""}, K::None, 0.1, t); + } + RunStep last; + for (int i = 0; i < 100 && last.action != RunAction::Finish; ++i) { + last = stepRun(run, ok, t); + if (last.action == RunAction::Poll) { + std::vector reps; + for (const auto& id : run.outstandingOpids()) reps.push_back(opRep(id, "failed", "proof failed")); + run.onPollResult(run.gen(), true, reps, t); + } + t += 0.5; + } + EXPECT_TRUE(run.end() == RunEnd::Halted); + EXPECT_FALSE(run.addresses()[3].done); + } + { // Transient: backoff 5, 10, 20, 40, 60 s, then the address fails. + ConsolidationRun run; + run.start({{"zA", 10, 5.0}}, RunConfig{}, 0.0); + double t = 0.0, last = 0.0; + const double expect[] = {5.0, 10.0, 20.0, 40.0, 60.0}; + for (int i = 0; i <= 5; ++i) { + RunStep s2; + do { s2 = stepRun(run, ok, t); if (s2.action != RunAction::Call) t += 0.25; } while (s2.action != RunAction::Call && t < 1000.0); + if (i > 0) EXPECT_TRUE(t - last >= expect[i - 1] - 0.01 && t - last <= expect[i - 1] + 0.3); + last = t; + run.onCallResult(run.gen(), 0, MergeCallResult{"", 0, 0, "HTTP 503"}, K::ShuttingDown, 0.0, t); + } + EXPECT_TRUE(run.addresses()[0].done && run.addresses()[0].outcome == AddrOutcome::Failed); + } + { // A result for another address or generation is an orphan: returned for tracking, state unchanged. + ConsolidationRun run; + run.start({{"zA", 10, 5.0}, {"zB", 10, 5.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + EXPECT_EQ(run.onCallResult(run.gen(), 1, MergeCallResult{"oX", 5, 5, ""}, K::None, 0.1, 0.1), std::string("oX")); + EXPECT_EQ(run.onCallResult(run.gen() + 7, 0, MergeCallResult{"oY", 5, 5, ""}, K::None, 0.1, 0.1), std::string("oY")); + EXPECT_TRUE(run.batches().empty()); + EXPECT_TRUE(stepRun(run, ok, 0.2).action == RunAction::None); // the real call is still in flight + } + { // Environment gates: Node and Busy hold calls and polls; Syncing, Locked, UserSend and Scan hold only calls. + ConsolidationRun run; + run.start({{"zA", 200, 100.0}, {"zB", 10, 5.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + run.onCallResult(run.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + RunEnv node = ok; node.nodeUsable = false; + RunEnv busy = ok; busy.busy = true; + EXPECT_TRUE(stepRun(run, node, 1.0).action == RunAction::None && run.lastWait() == RunWait::Node); + EXPECT_TRUE(stepRun(run, busy, 1.0).action == RunAction::None && run.lastWait() == RunWait::Busy); + RunEnv syncing = ok; syncing.callsAllowed = false; + EXPECT_TRUE(stepRun(run, syncing, 1.0).action == RunAction::Poll); // a due poll still runs + run.onPollResult(run.gen(), true, {opRep("o1", "executing")}, 1.0); + EXPECT_TRUE(stepRun(run, syncing, 3.0).action == RunAction::None && run.lastWait() == RunWait::Syncing); + RunEnv locked = ok; locked.locked = true; + EXPECT_TRUE(stepRun(run, locked, 3.0).action == RunAction::None && run.lastWait() == RunWait::Locked); + RunEnv send = ok; send.userSendBusy = true; + EXPECT_TRUE(stepRun(run, send, 3.0).action == RunAction::None && run.lastWait() == RunWait::UserSend); + RunEnv scan = ok; scan.scanBusy = true; + EXPECT_TRUE(stepRun(run, scan, 3.0).action == RunAction::None && run.lastWait() == RunWait::Scan); + EXPECT_TRUE(stepRun(run, ok, 3.0).action == RunAction::Call); + } + { // Stop: the call in flight is recorded, no further call, polls until terminal, then Stopped. + ConsolidationRun run; + run.start({{"zA", 200, 100.0}, {"zB", 10, 5.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + run.requestStop(); + run.onCallResult(run.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + double t = 0.1; + RunStep s2; + for (int i = 0; i < 100 && s2.action != RunAction::Finish; ++i) { + t += 0.5; + s2 = stepRun(run, ok, t); + EXPECT_TRUE(s2.action != RunAction::Call); + if (s2.action == RunAction::Poll) run.onPollResult(run.gen(), true, {opRep("o1", i < 4 ? "executing" : "success")}, t); + } + EXPECT_TRUE(run.end() == RunEnd::Stopped); + ConsolidationRun idle; // nothing outstanding, node gone: finishes at once + idle.start({{"zA", 10, 5.0}}, RunConfig{}, 0.0); + idle.requestStop(); + RunEnv gone = ok; gone.nodeUsable = false; + EXPECT_TRUE(idle.next(gone, 1.0).action == RunAction::Finish); + } + { // abort(): false unless running; the outstanding batch is Lost; late results are orphans / ignored. + ConsolidationRun run; + EXPECT_FALSE(run.abort()); + run.start({{"zA", 200, 100.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + const auto gen = run.gen(); + run.onCallResult(gen, 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + stepRun(run, ok, 2.5); // poll in flight + EXPECT_TRUE(run.abort()); + EXPECT_TRUE(run.end() == RunEnd::Aborted && run.batches()[0].state == OpState::Lost); + EXPECT_TRUE(run.addresses()[0].outcome == AddrOutcome::Uncertain); + EXPECT_EQ(run.onCallResult(gen, 0, MergeCallResult{"o9", 5, 5, ""}, K::None, 0.1, 3.0), std::string("o9")); + run.onPollResult(gen, true, {opRep("o1", "success")}, 3.0); + EXPECT_TRUE(run.batches()[0].state == OpState::Lost); + EXPECT_TRUE(run.next(ok, 4.0).action == RunAction::None); + EXPECT_FALSE(run.abort()); + } + { // Watchdogs: a call or poll outstanding 30 min is presumed lost; an op not terminal after an hour too. + ConsolidationRun run; + run.start({{"zA", 200, 100.0}, {"zB", 10, 5.0}}, RunConfig{}, 0.0); + stepRun(run, ok, 0.0); + const auto gen = run.gen(); + stepRun(run, ok, 1801.0); + EXPECT_TRUE(run.addresses()[0].done && run.addresses()[0].outcome == AddrOutcome::Uncertain); + EXPECT_EQ(run.gen(), gen + 1); + EXPECT_EQ(run.onCallResult(gen, 0, MergeCallResult{"late", 5, 5, ""}, K::None, 0.1, 1802.0), std::string("late")); + ConsolidationRun lost; + lost.start({{"zA", 200, 100.0}}, RunConfig{}, 0.0); + stepRun(lost, ok, 0.0); + lost.onCallResult(lost.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + stepRun(lost, ok, 0.2); + lost.onPollResult(lost.gen(), true, {}, 0.3); // missing once + EXPECT_TRUE(lost.batches()[0].state == OpState::Unknown); + stepRun(lost, ok, 3.5); + lost.onPollResult(lost.gen(), true, {}, 3.6); // twice: Lost + EXPECT_TRUE(lost.batches()[0].state == OpState::Lost); + ConsolidationRun hour; + hour.start({{"zA", 200, 100.0}}, RunConfig{}, 0.0); + stepRun(hour, ok, 0.0); + hour.onCallResult(hour.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + for (double t = 0.5; t < 3610.0; t += 30.0) { + const auto s2 = stepRun(hour, ok, t); + if (s2.action == RunAction::Poll) hour.onPollResult(hour.gen(), false, {}, t + 0.1); + } + EXPECT_TRUE(hour.batches()[0].state == OpState::Lost); + } + { // The op backstop counts only time in which a poll could have gone out (not while the node was gone or + // busy), with a 4 h cap whatever happened. + ConsolidationRun gated; + gated.start({{"zA", 200, 100.0}}, RunConfig{}, 0.0); + stepRun(gated, ok, 0.0); + gated.onCallResult(gated.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + RunEnv gone = ok; + gone.nodeUsable = false; + for (double t = 1.0; t < 7200.0; t += 30.0) stepRun(gated, gone, t); + EXPECT_FALSE(opStateTerminal(gated.batches()[0].state)); + EXPECT_TRUE(stepRun(gated, ok, 7201.0).action == RunAction::Poll); + gated.onPollResult(gated.gen(), true, {opRep("o1", "success")}, 7201.1); + EXPECT_TRUE(gated.batches()[0].state == OpState::Success); + ConsolidationRun capped; + capped.start({{"zA", 200, 100.0}}, RunConfig{}, 0.0); + stepRun(capped, ok, 0.0); + capped.onCallResult(capped.gen(), 0, MergeCallResult{"o1", 50, 150, ""}, K::None, 0.1, 0.1); + RunEnv busy = ok; + busy.busy = true; + for (double t = 1.0; t < 4 * 3600.0 + 60.0; t += 60.0) stepRun(capped, busy, t); + EXPECT_TRUE(capped.batches()[0].state == OpState::Lost); + } + { // The estimate is capped by maxBatches. + ConsolidationRun run; + run.start({{"zA", 10000, 5000.0}}, RunConfig{}, 0.0); + EXPECT_EQ(run.progress().batchesEstimated, 20); + } +} + +// A whole run against a deterministic fake daemon: serial ops, merge calls that cost 6 s, and windows of lock, +// catch-up and another wallet scan. +void testConsolidationRunSimulation() +{ + using namespace dragonx::services; + struct FakeOp { std::string id; std::string addr; int notes; double startAt = -1, finishAt = -1; }; + std::map unlocked{{"A", 240}, {"B", 60}, {"C", 3}, {"D", 1}}; + std::vector ops; + ConsolidationRun run; + std::vector groups; + for (const auto& [a, n] : unlocked) groups.push_back({a, n, 0.5 * n}); + EXPECT_TRUE(run.start(groups, RunConfig{}, 0.0)); + EXPECT_EQ(run.addresses().size(), static_cast(3)); // D (one note) is left out + int expected = 0; + for (const auto& a : run.addresses()) expected += estimateBatches(a.group.notes, 50); + // In-flight RPC model. + enum class Rpc { None, Call, Poll } rpc = Rpc::None; + double rpcDoneAt = 0.0, lastCallEnd = -1e9; + int rpcAddr = -1; + bool overlap = false, gapOk = true, windowOk = true, opsOk = true; + auto inWindow = [](double t, double a, double b) { return t >= a && t < b; }; + auto settle = [&](double t) { // the daemon's serial op queue + double prevFinish = 0.0; + for (auto& op : ops) { + if (op.startAt < 0) op.startAt = std::max(t, prevFinish); // set when queued below + prevFinish = op.finishAt; + } + }; + (void)settle; + auto status = [&](const FakeOp& op, double t) { + if (t < op.startAt) return std::string("queued"); + if (t < op.finishAt) return std::string("executing"); + return std::string("success"); + }; + double t = 0.0; + for (; t < 5000.0 && run.hasRun() && run.end() == RunEnd::None; t += 0.5) { + // Replies land. + if (rpc != Rpc::None && t >= rpcDoneAt) { + if (rpc == Rpc::Call) { + const std::string addr = run.addresses()[rpcAddr].group.address; + MergeCallResult r; + const int avail = unlocked[addr]; + if (avail < 2) { + r.error = "Could not find any funds to merge."; + run.onCallResult(run.gen(), rpcAddr, r, dragonx::rpc::RpcFailureKind::AppError, 6.0, t); + } else { + const int merging = std::min(50, avail); + unlocked[addr] -= merging; // the op locks its notes before the call returns + if (unlocked[addr] < 0) opsOk = false; + FakeOp op; + op.id = "op" + std::to_string(ops.size()); + op.addr = addr; + op.notes = merging; + const double prev = ops.empty() ? 0.0 : ops.back().finishAt; + op.startAt = std::max(t, prev); + op.finishAt = op.startAt + 8.0; + ops.push_back(op); + r.opid = op.id; + r.merging = merging; + r.remaining = unlocked[addr]; + run.onCallResult(run.gen(), rpcAddr, r, dragonx::rpc::RpcFailureKind::None, 6.0, t); + } + lastCallEnd = t; + } else { + std::vector reps; + for (const auto& id : run.outstandingOpids()) { + for (const auto& op : ops) if (op.id == id) reps.push_back(opRep(id, status(op, t))); + } + run.onPollResult(run.gen(), true, reps, t); + } + rpc = Rpc::None; + } + int unfinished = 0; + for (const auto& op : ops) if (status(op, t) != "success") ++unfinished; + if (unfinished > 2) opsOk = false; + RunEnv env = okRunEnv(); + env.locked = inWindow(t, 100.0, 160.0); + env.callsAllowed = !inWindow(t, 300.0, 330.0); + env.scanBusy = inWindow(t, 400.0, 410.0); + const RunStep st = run.next(env, t); + if (st.action == RunAction::Call) { + if (rpc != Rpc::None) overlap = true; + if (t - lastCallEnd < 6.0 - 1e-9) gapOk = false; + if (inWindow(t, 100.0, 160.0) || inWindow(t, 300.0, 330.0) || inWindow(t, 400.0, 410.0)) windowOk = false; + run.markCallPosted(st.addrIdx, t); + rpc = Rpc::Call; + rpcAddr = st.addrIdx; + rpcDoneAt = t + 6.0; + } else if (st.action == RunAction::Poll) { + if (rpc != Rpc::None) overlap = true; + run.markPollPosted(t); + rpc = Rpc::Poll; + rpcDoneAt = t + 0.1; + } + } + EXPECT_TRUE(run.end() == RunEnd::Completed); + EXPECT_FALSE(overlap); + EXPECT_TRUE(gapOk); + EXPECT_TRUE(windowOk); + EXPECT_TRUE(opsOk); + const auto sum = run.summary(); + EXPECT_EQ(sum.batchesSubmitted, expected); + EXPECT_EQ(expected, 8); + for (const auto& a : run.addresses()) EXPECT_TRUE(a.run.remaining < 2); + EXPECT_TRUE(runVerdict(sum) == RunVerdict::Clean); +} + int main() { testConnectionConfig(); @@ -11672,6 +12283,11 @@ int main() testDebugLogTailRules(); testNoteConsolidation(); testScopeScanAndLockedNotes(); + testMergeCallResult(); + testOpReports(); + testMergeFailureClassification(); + testConsolidationRun(); + testConsolidationRunSimulation(); testStopDeliveryPolicy(); testProcStatParse(); testStopProgressTracker(); diff --git a/tools/fake_dragonxd/README.md b/tools/fake_dragonxd/README.md index 07401cb..41101d4 100644 --- a/tools/fake_dragonxd/README.md +++ b/tools/fake_dragonxd/README.md @@ -122,6 +122,9 @@ inherits the GUI's environment. | `FAKE_STUCK_SHUTDOWN` | 0 | the Sep 29 orphan: hold the datadir lock, no listener, ignore SIGTERM, never exit | | `FAKE_WALLET_NOTES` | (off) | a note wallet: `addr:count:amount,...` shielded notes; enables `z_mergetoaddress` + its operations (dragonxd rules: confirmed unlocked notes, a batch locked on return, merged note next block) | | `FAKE_MERGE_SECS` | 6 | how long a `z_mergetoaddress` operation runs before it succeeds | +| `FAKE_MERGE_SERIAL` | 0 | `1` = ops run one at a time, like dragonxd's single async worker: a new op is `queued` until the previous one finishes | +| `FAKE_MERGE_FAIL` | (off) | `syncing:N` (the Nth and later merges fail with the THROW_IF_SYNCING text), `lockedkey:N` (-8 "Could not find spending key…" after the scan), `opfail:N` (the Nth and later ops end `failed`, their notes unlocked) | +| `FAKE_SCAN_SECS` | 0 | the GUI's wallet scans (`z_listunspent`, `listunspent`, `z_gettotalbalance`, `z_listreceivedbyaddress`, `z_mergetoaddress`) hold cs_main this long; `scan-begin`/`scan-end` events | | `FAKE_TIP_UNKNOWN_SECS` | 0 | `longestchain` reads 0 this long after `Done Loading` (a node that hasn't seen an UpdateTip yet) | | `FAKE_LONGEST_AHEAD` / `FAKE_CATCHUP_SECS` | 0 / 0 | the network tip (`longestchain`, `headers`) is N blocks ahead, for that long after `Done Loading` (0 = for good) | | `FAKE_TXCOUNT` | 42 | `getwalletinfo` `txcount` (the GUI's connect gate waits for the tip on wallets of 500+) | diff --git a/tools/fake_dragonxd/fake_dragonxd.cpp b/tools/fake_dragonxd/fake_dragonxd.cpp index 5769e3e..824a778 100644 --- a/tools/fake_dragonxd/fake_dragonxd.cpp +++ b/tools/fake_dragonxd/fake_dragonxd.cpp @@ -424,8 +424,10 @@ struct FakeOp { std::vector noteIds; std::string to; double fee = 0.0; + double startAt = 0.0; // FAKE_MERGE_SERIAL: queued until the previous op finished double finishAt = 0.0; // unixMs()/1000.0 std::string status = "executing"; + std::string error; // a failed op's message (FAKE_MERGE_FAIL=opfail:N) long long created = 0; }; bool g_noteWallet = false; @@ -434,6 +436,12 @@ std::vector g_ops; long g_noteSeq = 0; long g_opSeq = 0; double g_mergeSecs = 6.0; +// Consolidation-run knobs (C210 manual checks). +double g_scanSecs = 0.0; // FAKE_SCAN_SECS: a wallet scan holds cs_main this long +bool g_mergeSerial = false; // FAKE_MERGE_SERIAL=1: one op at a time, like dragonxd's single async worker +std::string g_mergeFailKind; // FAKE_MERGE_FAIL=syncing:N | lockedkey:N | opfail:N +long g_mergeFailFrom = 0; +long g_mergeCalls = 0; void loadNoteWallet() { @@ -441,6 +449,13 @@ void loadNoteWallet() if (!spec || !*spec) return; g_noteWallet = true; g_mergeSecs = envDouble("FAKE_MERGE_SECS", 6.0); + g_mergeSerial = envLong("FAKE_MERGE_SERIAL", 0) == 1; + if (const char* f = std::getenv("FAKE_MERGE_FAIL")) { + const std::string spec(f); + const size_t colon = spec.find(':'); + g_mergeFailKind = spec.substr(0, colon); + g_mergeFailFrom = colon == std::string::npos ? 1 : std::max(1L, std::strtol(spec.c_str() + colon + 1, nullptr, 10)); + } std::string s(spec); size_t pos = 0; while (pos <= s.size()) { @@ -472,7 +487,15 @@ void settleOps() { const double now = unixMs() / 1000.0; for (auto& op : g_ops) { + if (op.status == "queued" && now >= op.startAt) op.status = "executing"; if (op.status != "executing" || now < op.finishAt) continue; + if (!op.error.empty()) { // FAKE_MERGE_FAIL=opfail: the op fails and its notes are unlocked again + for (auto& n : g_notes) + if (std::find(op.noteIds.begin(), op.noteIds.end(), n.id) != op.noteIds.end()) n.locked = false; + op.status = "failed"; + event("merge-failed", op.id); + continue; + } double sum = 0.0; for (long id : op.noteIds) { auto it = std::find_if(g_notes.begin(), g_notes.end(), [&](const FakeNote& n) { return n.id == id; }); @@ -488,6 +511,7 @@ json opStatus(const FakeOp& op) { json o = {{"id", op.id}, {"status", op.status}, {"creation_time", op.created}, {"method", "z_mergetoaddress"}}; if (op.status == "success") o["result"] = {{"txid", fakeHash(1000000 + std::stol(op.id.substr(5)))}}; + if (op.status == "failed") o["error"] = {{"code", -6}, {"message", op.error}}; return o; } @@ -503,7 +527,7 @@ Canned noteWalletReply(const std::string& m, const json& params) const long conf = std::max(0L, h - n.height + 1); arr.push_back({{"txid", fakeHash(5000000 + n.id)}, {"outindex", 0}, {"confirmations", conf}, {"spendable", true}, {"address", n.address}, {"amount", n.amount}, {"memo", "f6"}, - {"change", false}}); + {"change", false}, {"locked", n.locked}}); } c.result = arr; } else if (m == "z_listaddresses") { @@ -535,6 +559,24 @@ Canned noteWalletReply(const std::string& m, const json& params) else from.push_back(a.get()); } const std::string to = params[1].get(); + ++g_mergeCalls; + const bool failNow = !g_mergeFailKind.empty() && g_mergeCalls >= g_mergeFailFrom; + if (failNow && g_mergeFailKind == "syncing") { // THROW_IF_SYNCING runs before the scan (cheap) + c.errCode = -1; + c.errMsg = "z_mergetoaddress: Extreme Privacy! Chain still syncing at height " + std::to_string(h) + + ", aborting to prevent linkability analysis. Please wait until FULLY SYNCED and try again."; + return c; + } + if (g_scanSecs > 0.0) { // GetFilteredNotes: a whole-wallet scan under cs_main + event("scan-begin", "m=z_mergetoaddress"); + std::this_thread::sleep_for(std::chrono::duration(g_scanSecs)); + event("scan-end", "m=z_mergetoaddress"); + } + if (failNow && g_mergeFailKind == "lockedkey") { + c.errCode = -8; + c.errMsg = "Could not find spending key for payment address."; + return c; + } const double fee = params.size() > 2 && params[2].is_number() ? params[2].get() : 0.0001; long limit = params.size() > 4 && params[4].is_number() ? params[4].get() : 90; if (limit <= 0) limit = 1000000; @@ -565,7 +607,16 @@ Canned noteWalletReply(const std::string& m, const json& params) op.to = to; op.fee = fee; op.created = unixMs() / 1000; - op.finishAt = unixMs() / 1000.0 + g_mergeSecs; + const double nowSec = unixMs() / 1000.0; + double prevFinish = nowSec; + if (g_mergeSerial) { + for (const auto& o : g_ops) + if (o.status == "queued" || o.status == "executing") prevFinish = std::max(prevFinish, o.finishAt); + } + op.startAt = prevFinish; + op.finishAt = op.startAt + g_mergeSecs; + op.status = op.startAt > nowSec ? "queued" : "executing"; + if (failNow && g_mergeFailKind == "opfail") op.error = "Proof failed (FAKE_MERGE_FAIL)"; g_ops.push_back(op); event("merge", op.id + " notes=" + std::to_string(picked.size()) + " to=" + to); c.result = {{"remainingUTXOs", 0}, {"remainingTransparentValue", 0.0}, @@ -581,7 +632,7 @@ Canned noteWalletReply(const std::string& m, const json& params) std::vector keep; for (const auto& op : g_ops) { const bool match = filter.empty() || std::find(filter.begin(), filter.end(), op.id) != filter.end(); - const bool finished = op.status != "executing"; + const bool finished = op.status == "success" || op.status == "failed"; if (match && (m == "z_getoperationstatus" || finished)) arr.push_back(opStatus(op)); if (!(m == "z_getoperationresult" && match && finished)) keep.push_back(op); } @@ -792,6 +843,13 @@ json execOne(const json& req, int& status) std::lock_guard lk(g_csMain); c = cannedReply(m, params); if (m == "setgenerate") g_minerOn = !params.empty() && params[0].is_boolean() && params[0].get(); + // FAKE_SCAN_SECS: the GUI's wallet-wide scans hold cs_main (z_mergetoaddress sleeps in its own handler). + if (g_scanSecs > 0.0 && (m == "z_listunspent" || m == "listunspent" || m == "z_gettotalbalance" || + m == "z_listreceivedbyaddress")) { + event("scan-begin", "m=" + m); + std::this_thread::sleep_for(std::chrono::duration(g_scanSecs)); + event("scan-end", "m=" + m); + } if (g_noteWallet) { Canned w = noteWalletReply(m, params); if (w.known) c = w; @@ -1363,6 +1421,7 @@ void loadConfig(int argc, char** argv) g_cfg.tipUnknownSecs = envDouble("FAKE_TIP_UNKNOWN_SECS", g_cfg.tipUnknownSecs); g_cfg.longestAhead = std::max(0L, envLong("FAKE_LONGEST_AHEAD", g_cfg.longestAhead)); g_cfg.catchupSecs = envDouble("FAKE_CATCHUP_SECS", g_cfg.catchupSecs); + g_scanSecs = envDouble("FAKE_SCAN_SECS", 0.0); g_cfg.txcount = envLong("FAKE_TXCOUNT", g_cfg.txcount); g_cfg.height = envLong("FAKE_HEIGHT", g_cfg.height); g_cfg.rebuildTotal = std::max(1L, envLong("FAKE_REBUILD_TOTAL", g_cfg.rebuildTotal));