diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 87608701..f4f09a42 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -79,7 +79,7 @@ jobs: strategy: fail-fast: false matrix: - config: [default, orchard, transparent, transparent-import, one-time-ivk, sqlite, enhance-wallet, enhance, transparent-pir, transparent-pir-sqlite, transparent-pir-testing] + config: [default, orchard, transparent, transparent-import, one-time-ivk, sqlite, enhance-wallet, enhance, transparent-pir, transparent-pir-sqlite, transparent-pir-testing, pir-receiver] env: WALLET_LIB_BUILD_ROOT: ${{ github.workspace }}/target/dev-lanes steps: diff --git a/Cargo.lock b/Cargo.lock index 899e9647..249df583 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -159,7 +159,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -170,7 +170,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1483,7 +1483,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1698,7 +1698,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -3262,7 +3262,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -3717,10 +3717,12 @@ name = "pir-control" version = "0.1.0" source = "git+https://github.com/valargroup/wallet-pir?rev=9d2cbda0b4336258cfce76ca8ee77b095daa7fd7#9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" dependencies = [ + "axum", "hex", "rand 0.8.8", "serde", "sha2 0.10.9", + "tokio", ] [[package]] @@ -4068,7 +4070,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -4262,6 +4264,52 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "receiver-directory" +version = "0.0.1" +source = "git+https://github.com/valargroup/wallet-pir?rev=9d2cbda0b4336258cfce76ca8ee77b095daa7fd7#9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" +dependencies = [ + "bitcoin", + "incrementalmerkletree", + "serde", + "serde_json", + "sha2 0.10.9", + "thiserror 2.0.21", + "zakura-orchard", + "zcash_note_encryption", +] + +[[package]] +name = "receiver-pir" +version = "0.0.1" +source = "git+https://github.com/valargroup/wallet-pir?rev=9d2cbda0b4336258cfce76ca8ee77b095daa7fd7#9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" +dependencies = [ + "hex", + "ipir-sp 0.1.0-rc.6 (git+https://github.com/valargroup/ipir-sp.git?tag=v0.1.0-rc.6)", + "pir-native", + "rand 0.8.8", + "receiver-directory", + "serde", + "serde_json", + "sha2 0.10.9", + "thiserror 2.0.21", +] + +[[package]] +name = "receiver-pir-server" +version = "0.0.1" +source = "git+https://github.com/valargroup/wallet-pir?rev=9d2cbda0b4336258cfce76ca8ee77b095daa7fd7#9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" +dependencies = [ + "axum", + "hex", + "pir-control", + "pir-observability", + "receiver-directory", + "receiver-pir", + "serde_json", + "tokio", +] + [[package]] name = "reddsa" version = "0.5.2" @@ -4567,7 +4615,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -5134,7 +5182,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -5343,7 +5391,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7350,7 +7398,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7902,9 +7950,9 @@ dependencies = [ "zakura-address", "zakura-bls12-381", "zakura-client-backend", - "zakura-one-time-ivk", "zakura-jubjub", "zakura-keys", + "zakura-one-time-ivk", "zakura-orchard", "zakura-pasta-curves", "zakura-pczt", @@ -8183,6 +8231,33 @@ dependencies = [ "reinspiring 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "zakura-pir-receiver" +version = "0.0.1-rc0" +dependencies = [ + "axum", + "base64 0.22.1", + "futures-util", + "hex", + "receiver-directory", + "receiver-pir", + "receiver-pir-server", + "rusqlite", + "serde_json", + "sha2 0.10.9", + "thiserror 2.0.21", + "tokio", + "tower", + "zakura-client-backend", + "zakura-client-sqlite", + "zakura-one-time-ivk", + "zakura-orchard", + "zakura-pir-enhance", + "zakura-primitives", + "zakura-protocol", + "zcash_note_encryption", +] + [[package]] name = "zakura-pir-status" version = "0.0.1-rc0" diff --git a/Cargo.toml b/Cargo.toml index 5fe8f95b..39ffb47a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,7 @@ members = [ "zakura/transaction-status", "zakura/pir-transparent", "zakura/one-time-ivk", + "zakura/pir-receiver", "zakura/near-intents", ] diff --git a/README.md b/README.md index d523c71a..c78b45ee 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,9 @@ regeneration command, separate from ordinary builds. The [one-time IVK](zakura/one-time-ivk/README.md) crate provides shared derivation of receiving keys that keep an account's spending authority with their own incoming viewing keys, and refund memo helpers, with an Ironwood proof test for -mixed ordinary, refund, and incoming inputs. +mixed ordinary, refund, and incoming inputs. The unpublished +[receiver PIR](zakura/pir-receiver/README.md) crate runs the restore sweeps that +find payments to those keys through a receiver directory. ## How the rewiring works diff --git a/docs/development.md b/docs/development.md index de5b8866..60f00e29 100644 --- a/docs/development.md +++ b/docs/development.md @@ -18,7 +18,7 @@ python3 scripts/dev.py verify | Configuration | Packages and features | | --- | --- | -| `default` | Workspace except the facade, Enhance PIR and the transparent PIR adapter; default features | +| `default` | Workspace except the facade, Enhance PIR, the transparent PIR adapter and receiver PIR; default features | | `orchard` | Backend + SQLite; `orchard,test-dependencies` | | `transparent` | Backend + SQLite; `orchard,transparent-inputs,test-dependencies,unstable` | | `transparent-import` | Backend + SQLite; transparent features plus `transparent-key-import` | @@ -29,6 +29,7 @@ python3 scripts/dev.py verify | `transparent-pir` | Transparent PIR adapter; `wallet` | | `transparent-pir-sqlite` | Transparent PIR adapter; `sqlite` (`wallet` plus the SQLite apply-and-acknowledge integration and the in-process end-to-end tests) | | `transparent-pir-testing` | Transparent PIR adapter; `sqlite` and `testing` (the test support applications build their recovery and txid display tests on) | +| `pir-receiver` | Receiver PIR; `wallet` (which enables the backend's `one-time-ivk`; its tests use SQLite with `one-time-ivk`) | `default` builds the backend and SQLite crates without Orchard or transparent inputs, which `orchard` and `transparent` test, so items used only under those diff --git a/librustzcash/zcash_client_sqlite/src/wallet/one_time_ivk.rs b/librustzcash/zcash_client_sqlite/src/wallet/one_time_ivk.rs index 58dd28b5..531874da 100644 --- a/librustzcash/zcash_client_sqlite/src/wallet/one_time_ivk.rs +++ b/librustzcash/zcash_client_sqlite/src/wallet/one_time_ivk.rs @@ -26,8 +26,9 @@ //! [`WalletDb::begin_receive_operation`], [`WalletDb::finish_receive_operation`] and //! [`WalletDb::start_receive_operation`]. //! - [`WalletDb::record_operation_status`] for each provider status of either. -//! - The [`OneTimeIvkRestoreWrite`] sweep steps for restore sweeps, and -//! [`WalletDb::recheck_one_time_key_history`] when the user asks to recheck swaps. +//! - `zakura_pir_receiver::sweep`, over the [`OneTimeIvkRestoreWrite`] steps, for restore +//! sweeps, and [`WalletDb::recheck_one_time_key_history`] when the user asks to recheck +//! swaps. //! //! [`OneTimeIvkRestoreWrite`]: zcash_client_backend::data_api::one_time_ivk::restore::OneTimeIvkRestoreWrite //! [`OneTimeIvkRestoreWrite::maintain_one_time_ivks`]: zcash_client_backend::data_api::one_time_ivk::restore::OneTimeIvkRestoreWrite::maintain_one_time_ivks diff --git a/manifests/sources.toml b/manifests/sources.toml index 0d14ddbf..3f0d261c 100644 --- a/manifests/sources.toml +++ b/manifests/sources.toml @@ -27,7 +27,7 @@ vendored_directory = "librustzcash" vendor_branch = "vendor/librustzcash" upstream_manifest = "librustzcash/upstream-workspace.toml" # `zakura/` members are added here as they are written. -extra_members = ["wallet-lib", "zakura/pir-enhance", "zakura/pir-enhance-types", "zakura/pir-native", "zakura/pir-status", "zakura/transaction-status", "zakura/pir-transparent", "zakura/one-time-ivk", "zakura/near-intents"] +extra_members = ["wallet-lib", "zakura/pir-enhance", "zakura/pir-enhance-types", "zakura/pir-native", "zakura/pir-status", "zakura/transaction-status", "zakura/pir-transparent", "zakura/one-time-ivk", "zakura/pir-receiver", "zakura/near-intents"] # The backend selector. Named separately because the verification scripts have # to build it one backend at a time, unlike every other member. facade = "wallet-lib" diff --git a/scripts/dev.py b/scripts/dev.py index 79b2d4cb..ae26303a 100644 --- a/scripts/dev.py +++ b/scripts/dev.py @@ -15,7 +15,7 @@ ROOT = Path(__file__).resolve().parents[1] CONFIGS = { - "default": (["--workspace", "--exclude", "zakura-wallet-lib", "--exclude", "zakura-pir-enhance", "--exclude", "zakura-pir-transparent"], []), + "default": (["--workspace", "--exclude", "zakura-wallet-lib", "--exclude", "zakura-pir-enhance", "--exclude", "zakura-pir-transparent", "--exclude", "zakura-pir-receiver"], []), "orchard": (["-p", "zakura-client-backend", "-p", "zakura-client-sqlite"], ["orchard", "test-dependencies"]), "transparent": (["-p", "zakura-client-backend", "-p", "zakura-client-sqlite"], ["orchard", "transparent-inputs", "test-dependencies", "unstable"]), "transparent-import": (["-p", "zakura-client-backend", "-p", "zakura-client-sqlite"], ["orchard", "transparent-inputs", "transparent-key-import", "test-dependencies", "unstable"]), @@ -26,6 +26,7 @@ "transparent-pir": (["-p", "zakura-pir-transparent"], ["wallet"]), "transparent-pir-sqlite": (["-p", "zakura-pir-transparent"], ["sqlite"]), "transparent-pir-testing": (["-p", "zakura-pir-transparent"], ["sqlite", "testing"]), + "pir-receiver": (["-p", "zakura-pir-receiver"], ["wallet"]), } VERIFY = ("zakura-graph", "wallet-lib-modes", "vendor-ancestry") diff --git a/scripts/verify-zakura-graph.sh b/scripts/verify-zakura-graph.sh index c8db194e..2ca6a822 100755 --- a/scripts/verify-zakura-graph.sh +++ b/scripts/verify-zakura-graph.sh @@ -36,10 +36,11 @@ PY cargo check --manifest-path "$repo_root/Cargo.toml" \ --workspace --exclude "$facade" --all-targets --all-features --locked -# No development lane builds the transparent PIR adapter without `wallet`: its -# tests serve shards in process, so the `default` lane leaves it out. +# No development lane builds the transparent PIR adapter or the receiver PIR +# client without `wallet`: their tests serve shards or directories in process, +# so the `default` lane leaves them out. cargo check --manifest-path "$repo_root/Cargo.toml" \ - -p zakura-pir-transparent --lib --locked + -p zakura-pir-transparent -p zakura-pir-receiver --lib --locked cargo metadata --locked --manifest-path "$repo_root/Cargo.toml" \ --format-version 1 --all-features \ @@ -101,6 +102,16 @@ for name, found in sorted(versions.items()): if len(found) > 1 and (name.startswith("zakura-") or name in expected_packages): problems.append(f"{name}: {len(found)} versions in the graph: {sorted(found)}") +# Two git sources of one name are separate crates even at one version. Checked +# over the whole workspace graph, where the PIR clients share wallet-pir crates. +git_sources = defaultdict(set) +for package in packages.values(): + if (package["source"] or "").startswith("git+"): + git_sources[package["name"]].add(package["source"]) +for name, found in sorted(git_sources.items()): + if len(found) > 1: + problems.append(f"{name}: {len(found)} git sources in the graph: {sorted(found)}") + missing = expected_packages - versions.keys() for name in sorted(missing): problems.append(f"{name}: vendored crate is not in the resolved graph") diff --git a/zakura/one-time-ivk/README.md b/zakura/one-time-ivk/README.md index 5c244420..f6a42adc 100644 --- a/zakura/one-time-ivk/README.md +++ b/zakura/one-time-ivk/README.md @@ -23,9 +23,10 @@ issuing live addresses. The wallet owns reservation, persistence, coverage, lifecycle, and note selection. There is no alternate balance store in this crate. -The SQLite backend implements this contract with its `orchard` feature: durable -key registration, scanning, completion, and restore sweeps. Its rustdoc, starting -at `zcash_client_sqlite::wallet::one_time_ivk`, documents the calls a wallet makes. +The SQLite backend implements this contract with its `one-time-ivk` feature: +durable key registration, scanning, completion, and the restore sweeps that +`zakura-pir-receiver` runs. Its rustdoc, starting at +`zcash_client_sqlite::wallet::one_time_ivk`, documents the calls a wallet makes. Software PCZT signing of one-time-key notes works through the same builder. Hardware signers need firmware qualification before one-time keys are enabled for diff --git a/zakura/pir-receiver/CHANGELOG.md b/zakura/pir-receiver/CHANGELOG.md new file mode 100644 index 00000000..33ab571a --- /dev/null +++ b/zakura/pir-receiver/CHANGELOG.md @@ -0,0 +1,13 @@ +# Changelog + +## [Unreleased] + +### Added + +- `sweep`, behind the `wallet` feature, which runs one account's receiver-directory + restore sweeps of its one-time keys for any store implementing the backend's + `one_time_ivk::restore::OneTimeIvkRestoreWrite`, with `Transport`, `NoteSource`, + `EnhanceNotes`, `WriteLock`, `Swept`, `Error` and `MAINNET_GENESIS`. The receiver PIR + client moved here from Vizor. `wallet` enables the backend's `one-time-ivk` feature. +- `fetch_seen` and `Seen`: a publication's swap provider seen sets, which a wallet + checks before issuing a swap address. They need no `wallet` feature. diff --git a/zakura/pir-receiver/Cargo.toml b/zakura/pir-receiver/Cargo.toml new file mode 100644 index 00000000..ab17dbe6 --- /dev/null +++ b/zakura/pir-receiver/Cargo.toml @@ -0,0 +1,47 @@ +[package] +name = "zakura-pir-receiver" +version = "0.0.1-rc0" +publish = false +description = "Restore sweeps of one-time IVKs through a receiver directory over PIR" +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +readme = "README.md" + +[features] +default = [] +## The restore sweep, over a wallet store's one-time IVK traits. +wallet = ["dep:futures-util", "dep:thiserror", "dep:zakura-pir-enhance", "dep:zakura-one-time-ivk", "dep:zcash_client_backend", "dep:zcash_protocol"] + +[dependencies] +futures-util = { version = "0.3", optional = true } +thiserror = { version = "2", optional = true } +zakura-pir-enhance = { version = "=0.0.1-rc1", path = "../pir-enhance", features = ["wallet"], optional = true } +zakura-one-time-ivk = { version = "=0.0.1-rc0", path = "../one-time-ivk", optional = true } +zcash_client_backend = { workspace = true, features = ["one-time-ivk"], optional = true } +zcash_protocol = { workspace = true, optional = true } +# The receiver crates come from wallet-pir's `main`. +receiver-directory = { git = "https://github.com/valargroup/wallet-pir", rev = "9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" } +receiver-pir = { git = "https://github.com/valargroup/wallet-pir", rev = "9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" } + +[dev-dependencies] +axum = "0.7" +base64.workspace = true +hex.workspace = true +orchard.workspace = true +receiver-pir = { git = "https://github.com/valargroup/wallet-pir", rev = "9d2cbda0b4336258cfce76ca8ee77b095daa7fd7", features = ["server"] } +receiver-pir-server = { git = "https://github.com/valargroup/wallet-pir", rev = "9d2cbda0b4336258cfce76ca8ee77b095daa7fd7" } +rusqlite = { workspace = true, features = ["hooks"] } +serde_json.workspace = true +sha2.workspace = true +tokio = { workspace = true, features = ["macros", "rt"] } +tower = { workspace = true, features = ["util"] } +zcash_client_backend = { workspace = true, features = ["one-time-ivk", "test-dependencies"] } +zcash_client_sqlite = { path = "../../librustzcash/zcash_client_sqlite", package = "zakura-client-sqlite", features = ["one-time-ivk", "test-dependencies"] } +zcash_note_encryption.workspace = true +zcash_primitives.workspace = true + +[[test]] +name = "sweep" +required-features = ["wallet"] diff --git a/zakura/pir-receiver/README.md b/zakura/pir-receiver/README.md new file mode 100644 index 00000000..4b379007 --- /dev/null +++ b/zakura/pir-receiver/README.md @@ -0,0 +1,24 @@ +# Receiver directory restore sweeps + +A restored wallet finds payments to its one-time keys by looking each one up in a +receiver directory over PIR. `fetch_seen` downloads the swap provider seen sets that +a wallet checks before issuing a swap address. With the `wallet` feature, `sweep` +runs one account's restore sweeps for any wallet store that implements the backend's +restore traits (`zcash_client_backend::data_api::one_time_ivk::restore`, from its +`one-time-ivk` feature), such as `zakura-client-sqlite`. Its rustdoc states which +publications a run accepts, when it sends requests, which keys keep scanning or are +checked again, and how failures are retried. + +Applications supply: + +- the directory's HTTPS origin and a `Transport` for it, and a `NoteSource` for the + note data of found payments, usually `EnhanceNotes` over Enhance PIR. Both carry + the application's route policy, timeouts and cancellation; this crate has no HTTP + client. +- a `WriteLock` that serializes the sweep's wallet writes with the application's + other writers. +- the wallet's consensus parameters, the network's genesis hash (`MAINNET_GENESIS` on + mainnet) and the clock. + +`tests/sweep.rs` serves a directory in process through the receiver service's own +routes and real PIR, and checks each of those rules against restored SQLite wallets. diff --git a/zakura/pir-receiver/src/lib.rs b/zakura/pir-receiver/src/lib.rs new file mode 100644 index 00000000..a51ab013 --- /dev/null +++ b/zakura/pir-receiver/src/lib.rs @@ -0,0 +1,35 @@ +//! Restore sweeps of one-time keys through a receiver directory over PIR. +//! +//! A restored wallet cannot rescan public history for every key its seed may have +//! used, so each one-time key recovered from the seed is looked up once in a receiver +//! directory: a public index of Ironwood payments sent with the zero outgoing viewing +//! key, keyed by receiver. Lookups use PIR, so the directory does not learn which +//! receivers the wallet holds. A publication is used only at a block the wallet +//! scanned, and the wallet authenticates every payment with its derived key and its +//! own chain before crediting it. +//! +//! `fetch_seen` downloads the swap provider seen sets that a wallet checks before +//! issuing a swap address. The `wallet` feature adds `sweep`, which runs these lookups +//! for any wallet store that implements the backend's `OneTimeIvkRestoreWrite`. The +//! caller supplies transports, which carry its route policy, cancellation and timeouts, +//! and a `WriteLock` that serializes wallet writes with its other writers. Dropping a +//! `sweep` future at any await is safe: each wallet step commits on its own, and every +//! begun attempt is already backed off. + +mod seen; +#[cfg(feature = "wallet")] +mod sweep; +pub use seen::{Seen, fetch_seen}; +#[cfg(feature = "wallet")] +pub use sweep::{EnhanceNotes, Error, NoteSource, Swept, WriteLock, sweep}; + +pub use receiver_pir::{Error as DirectoryError, transport::Transport}; +/// The directory format and client underneath, for callers that need them directly. +pub use {receiver_directory, receiver_pir}; + +/// Zcash mainnet's genesis block hash, in internal byte order, which its directory +/// publications commit to. +pub const MAINNET_GENESIS: [u8; 32] = [ + 0x08, 0xce, 0x3d, 0x97, 0x31, 0xb0, 0x00, 0xc0, 0x83, 0x38, 0x45, 0x5c, 0x8a, 0x4a, 0x6b, 0xd0, + 0x5d, 0xa1, 0x6e, 0x26, 0xb1, 0x1d, 0xaa, 0x1b, 0x91, 0x71, 0x84, 0xec, 0xe8, 0x0f, 0x04, 0x00, +]; diff --git a/zakura/pir-receiver/src/seen.rs b/zakura/pir-receiver/src/seen.rs new file mode 100644 index 00000000..890f766f --- /dev/null +++ b/zakura/pir-receiver/src/seen.rs @@ -0,0 +1,116 @@ +//! Swap provider seen sets, which wallets read before issuing a swap address. +use std::collections::BTreeMap; + +use receiver_directory::{ + Receiver, + filter::{Filter, RECENT, SEEN}, + snapshot::FilterSet, +}; +use receiver_pir::transport::DirectoryClient; + +use crate::{DirectoryError, Transport}; + +/// The swap provider seen sets of one receiver directory publication: every receiver +/// each provider was given while its feed ran. Every wallet downloads them alike. +pub struct Seen { + salt: receiver_directory::Hash, + sets: Vec, + since: i64, + until: i64, +} + +impl Seen { + /// When the latest-starting provider feed started, in Unix seconds. Receivers given + /// earlier may be missing. + pub fn since(&self) -> i64 { + self.since + } + + /// When the earliest of the feeds' last complete reads began, in Unix seconds. + /// Receivers given later may be missing. + pub fn until(&self) -> i64 { + self.until + } + + /// Whether any provider's set holds each of `receivers`, in order. A set may hold a + /// receiver it was never given, at its filter's false positive rate. A receiver + /// that is not a valid Orchard-format receiver is reported as held. + pub fn contains(&self, receivers: &[[u8; 43]]) -> Vec { + let parsed: Vec<_> = receivers + .iter() + .map(|r| Receiver::from_bytes(*r).ok()) + .collect(); + let valid: Vec<_> = parsed.iter().flatten().copied().collect(); + let mut held = vec![false; valid.len()]; + for set in &self.sets { + for (held, hit) in held.iter_mut().zip(set.matches(&self.salt, &valid)) { + *held |= hit; + } + } + let mut held = held.into_iter(); + parsed + .iter() + .map(|r| r.is_none() || held.next().unwrap_or(true)) + .collect() + } +} + +/// Downloads the seen sets of the publication `origin` serves, which must commit to +/// the network's `genesis` ([`MAINNET_GENESIS`](crate::MAINNET_GENESIS) on mainnet), as +/// for `sweep`, since another network's sets say nothing of this one's quotes. Fails if +/// it declares no seen sets, or unless every provider with a recent or seen set has a +/// seen set dated with its feed's read times, since receivers a provider without one had +/// would look unseen. +pub async fn fetch_seen( + origin: &str, + transport: &T, + genesis: [u8; 32], +) -> Result { + let manifest = DirectoryClient::fetch_manifest(origin, transport).await?; + // The error `sweep`'s coverage check gives for another genesis. + if manifest.directory.genesis != genesis { + return Err(receiver_directory::Error::Coverage.into()); + } + let filters = DirectoryClient::fetch_filters(origin, transport, &manifest).await?; + let mut seen = Seen { + salt: manifest.directory.salt, + sets: Vec::new(), + since: i64::MIN, + until: i64::MAX, + }; + for (label, since, until) in seen_sets(&manifest.directory.filters)? { + seen.since = seen.since.max(since); + seen.until = seen.until.min(until); + // Fetching the filters checked them against the manifest's sets. + seen.sets + .push(filters.get(label).ok_or(DirectoryError::Malformed)?.clone()); + } + Ok(seen) +} + +/// The labels and feed read times of the seen sets among a publication's filter `sets`, +/// failing as [`fetch_seen`] does. +pub(crate) fn seen_sets(sets: &[FilterSet]) -> Result, DirectoryError> { + let mut providers = BTreeMap::<&str, Option<(&str, i64, i64)>>::new(); + for set in sets { + match set.label.split_once('/') { + Some((provider, RECENT)) => { + providers.entry(provider).or_default(); + } + Some((provider, SEEN)) => { + let (Some(since), Some(until)) = (set.since_unix, set.until_unix) else { + return Err(DirectoryError::Malformed); + }; + providers.insert(provider, Some((&set.label, since, until))); + } + _ => {} + } + } + if providers.is_empty() { + return Err(DirectoryError::Malformed); + } + providers + .into_values() + .map(|seen| seen.ok_or(DirectoryError::Malformed)) + .collect() +} diff --git a/zakura/pir-receiver/src/sweep.rs b/zakura/pir-receiver/src/sweep.rs new file mode 100644 index 00000000..b7f5525f --- /dev/null +++ b/zakura/pir-receiver/src/sweep.rs @@ -0,0 +1,696 @@ +//! Restore sweeps over a wallet's one-time IVK traits. + +use std::{collections::BTreeMap, num::NonZeroU32}; + +use futures_util::StreamExt; +use receiver_directory::{ + Receiver, + filter::{Filters, PAID, RECENT}, + snapshot::FilterSet, + witness::WitnessSnapshot, +}; +use receiver_pir::{AcceptedCoverage, transport::DirectoryClient}; +use zakura_one_time_ivk::{KeyId, swap::RESTORE_WATCH_SECS}; +use zakura_pir_enhance::{ + ClientError, ClientResourceLimits, + transport::{Client, PendingClient, Transport as EnhanceTransport}, + wallet::{Acceptance, acceptance}, +}; +use zcash_client_backend::data_api::{ + WalletRead, + enhance_pir::EnhancePirRead, + one_time_ivk::restore::{ + DirectoryPayment, DiscoveryWork, OneTimeIvkRestoreRead, OneTimeIvkRestoreWrite, + PaymentApplication, SweepDeferral, + }, + transparent_ledger::ChainPoint, +}; +use zcash_protocol::consensus::{BlockHeight, NetworkUpgrade, Parameters}; + +use crate::{DirectoryError, Transport, seen::seen_sets}; + +/// A wallet's account identifier. +type AccountId = ::AccountId; +/// A wallet's error type. +type AccountError = ::Error; + +/// Due sweeps prepared per run. +const BATCH: NonZeroU32 = NonZeroU32::new(64).unwrap(); +/// Positions whose note data one request asks for. Each batch is queued as it arrives, +/// so a long history keeps its progress when a run stops partway. +const NOTE_BATCH: usize = 64; +/// How long before `now` a provider's recent set may have last read its feed and still +/// rule a receiver out. A quote made after that read is missing from the set. +const RECENT_SET_MAX_AGE_SECS: i64 = 15 * 60; + +/// Why a sweep run, or one key's sweep, stopped. `E` is the wallet's error type. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The receiver directory refused, failed or answered inconsistently. + #[error("receiver directory: {0}")] + Directory(#[from] DirectoryError), + /// Enhance PIR could not supply note data. + #[error("note data: {0}")] + NoteData(#[from] ClientError), + /// A wallet step refused the work, or the wallet's storage failed. + #[error(transparent)] + Wallet(E), + /// A wallet step must wait for more scanning or a newer publication. + #[error("deferred: {0:?}")] + Deferred(SweepDeferral), + /// A service's publication cannot be used against the wallet's chain yet. + #[error("{0}")] + Unavailable(&'static str), + /// The note source cannot serve the wallet's chain, so no key can get note data + /// this run. + #[error("note source: {0}")] + NotesUnavailable(&'static str), + /// A queued payment is waiting, for example for its witness. + #[error("payment remains queued: {0:?}")] + Queued(PaymentApplication), +} + +/// Serializes a sweep's wallet writes with the caller's other writers. +pub trait WriteLock { + /// Runs `write` while holding the caller's wallet write lock. `label` names the + /// step. + fn write(&self, label: &'static str, write: impl FnOnce() -> T) -> T; +} + +/// Supplies the note data that payments a lookup found need. +#[allow(async_fn_in_trait)] +pub trait NoteSource { + /// The 528 ciphertext bytes after the directory's prefix for each of `positions`, + /// keyed by commitment tree position. `wallet` is the wallet being swept. + /// `positions` is never empty when a sweep calls this. [`Error::NotesUnavailable`] + /// stops the run (see [`sweep`]). + async fn note_data( + &mut self, + wallet: &W, + positions: Vec, + ) -> Result, Error>>; +} + +/// Note data over Enhance PIR. Its session opens on first use, accepted against the +/// wallet's scanned chain, so a run whose lookups find nothing makes no Enhance +/// request. A later batch first accepts fresh routing if Enhance says a refresh is +/// due: thirty seconds on, or after a 409 or 410. A failed batch is not retried; the +/// next call refreshes. +pub struct EnhanceNotes<'a, T, P> { + origin: &'a str, + transport: &'a T, + params: &'a P, + session: Option, +} + +impl<'a, T: EnhanceTransport, P: Parameters> EnhanceNotes<'a, T, P> { + /// Note data from the Enhance PIR service at `origin`, reached through `transport`, + /// for a wallet on `params`' network. + pub fn new(origin: &'a str, transport: &'a T, params: &'a P) -> Self { + Self { + origin, + transport, + params, + session: None, + } + } +} + +impl NoteSource for EnhanceNotes<'_, T, P> +where + T: EnhanceTransport, + P: Parameters, + W: OneTimeIvkRestoreRead + EnhancePirRead, +{ + async fn note_data( + &mut self, + wallet: &W, + positions: Vec, + ) -> Result, Error>> { + let mut note_data = BTreeMap::new(); + if positions.is_empty() { + return Ok(note_data); + } + if self.session.as_ref().is_none_or(Client::refresh_due) { + let pending = PendingClient::fetch(self.transport, self.origin).await?; + let limits = ClientResourceLimits::with_cache(32768, 2); + let accepted = acceptance(wallet, pending.manifest(), self.params, limits) + .map_err(Error::Wallet)??; + let accepted = match accepted { + Acceptance::Accepted(accepted) => accepted, + Acceptance::WaitingForScanning => { + return Err(Error::NotesUnavailable( + "the Enhance PIR anchor has not been scanned", + )); + } + Acceptance::Mismatch => { + return Err(Error::NotesUnavailable( + "the Enhance PIR anchor differs from the wallet's chain", + )); + } + }; + match &mut self.session { + Some(session) => session.accept_routing(pending, &accepted)?, + None => self.session = Some(pending.accept(&accepted)?), + } + } + let session = self.session.as_mut().expect("opened above"); + // Enhance groups these positions into shared row requests itself. + let stream = session.query_batch(self.transport, positions)?; + futures_util::pin_mut!(stream); + while let Some(result) = stream.next().await { + note_data.insert(result.position, *result.record?.enc_ciphertext_suffix()); + } + Ok(note_data) + } +} + +/// What one [`sweep`] run did. `E` is the wallet's error type. +#[derive(Debug)] +pub struct Swept { + /// Keys whose sweep finished. + pub finished: usize, + /// Keys left for a later run, already backed off, with why. + pub deferred: Vec<(KeyId, Error)>, + /// Whether the account still has unfinished sweeps or queued payments. + pub pending: bool, +} + +/// Runs `account`'s due restore sweeps (see [`OneTimeIvkRestoreWrite`]) against the +/// receiver directory at `origin`, releases the spend evidence they no longer need (see +/// [`OneTimeIvkRestoreWrite::finish_one_time_nullifier_recovery`]), and checks closed keys +/// against each newer publication's paid filter (see +/// [`OneTimeIvkRestoreWrite::record_paid_filter_check`]). A wallet with several accounts +/// calls it once for each. +/// +/// Nothing runs unless `through`, the wallet's fully scanned tip, is also its stored chain +/// tip and the block it scanned there, which release checks again under its write lock +/// with the fully scanned height. A run with nothing due or to check makes no request. +/// Otherwise the publication must commit to `genesis` +/// ([`MAINNET_GENESIS`](crate::MAINNET_GENESIS) on mainnet), cover history from Ironwood +/// activation on `params`' network, end at an anchor the wallet accepts (see +/// [`OneTimeIvkRestoreRead::directory_publication_anchor`]) with the wallet's Ironwood +/// tree size there, and have the seen sets [`fetch_seen`](crate::fetch_seen) requires. +/// +/// Only receivers in the paid filter are looked up over PIR, with note data from `notes`. +/// The seen and recent sets give each key's `seen` and `recent` (see +/// [`OneTimeIvkRestoreWrite::begin_one_time_sweep_attempt`] and +/// [`OneTimeIvkRestoreWrite::apply_one_time_sweep`]). A provider's recent set counts only +/// if its feed's last read began at most fifteen minutes before `now` and its window, with +/// the feed running throughout, covers [`RESTORE_WATCH_SECS`]; unless every provider has +/// one, every key is recent. +/// +/// A run sweeps at most one batch of due keys, then moves the incoming lookahead, leaving +/// its new keys to a later run ([`Swept::pending`]). A key that fails is backed off and +/// listed in [`Swept::deferred`] while the others go on, unless the directory's session or +/// connection failed or `notes` cannot serve the wallet's chain: then the run stops leasing +/// keys. A failed wallet write stops the run with its error. `now` is the caller's clock in +/// seconds. +#[allow(clippy::too_many_arguments)] +pub async fn sweep( + wallet: &mut W, + params: &P, + account: AccountId, + through: ChainPoint, + genesis: [u8; 32], + origin: &str, + transport: T, + notes: &mut N, + lock: &L, + now: i64, +) -> Result>, Error>> +where + W: OneTimeIvkRestoreWrite, + P: Parameters, + T: Transport, + N: NoteSource, + L: WriteLock, +{ + let mut finished = 0; + let mut deferred = Vec::new(); + let current = at_tip(wallet, through).map_err(Error::Wallet)?; + let mut transport = Some(transport); + let mut directory = None; + if current { + let work = lock + .write("one_time_sweep.due", || { + wallet.prepare_one_time_sweeps(account, through, now, BATCH) + }) + .map_err(Error::Wallet)? + .map_err(Error::Deferred)?; + if !work.is_empty() { + let transport = transport.take().expect("not yet used"); + let accepted = + Directory::accept(wallet, params, through, genesis, origin, transport).await; + let directory = directory.insert(match accepted { + Ok(directory) => directory, + Err(e) => { + // No key can be swept without the directory, so each backs off as + // if its own lookup had failed, and the next sync does not wait on + // the same directory again. + for item in &work { + lock.write("one_time_sweep.attempt", || { + wallet.begin_one_time_sweep_attempt(account, item.key, now, false) + }) + .map_err(Error::Wallet)?; + } + return Err(e); + } + }); + let checks = directory.check(&work, now).await?; + let lookups = checks.iter().filter(|check| check.paid).count(); + if let Some(session) = &mut directory.session { + session.client.use_file_for_work(lookups).await?; + } + for (item, check) in work.into_iter().zip(checks) { + // A key whose lease is not stored would be due again at once. + lock.write("one_time_sweep.attempt", || { + wallet.begin_one_time_sweep_attempt(account, item.key, now, check.seen) + }) + .map_err(Error::Wallet)?; + let swept_key = sweep_key( + wallet, directory, account, &item, check, lookups, through, notes, lock, + ); + match swept_key.await { + Ok(()) => finished += 1, + Err(e) => { + // Every later key would fail the same way and back off. A session + // that failed to open took the transport without replacing it. + let unavailable = directory.transport.is_none() + && directory.session.is_none() + || matches!( + e, + Error::Directory( + DirectoryError::Revision | DirectoryError::Transport(_) + ) | Error::NotesUnavailable(_) + ); + deferred.push((item.key, e)); + if unavailable { + break; + } + } + } + } + // New lookahead keys wait for a later run, which bounds this one. + lock.write("one_time_sweep.lookahead", || { + wallet.maintain_one_time_ivks(account) + }) + .map_err(Error::Wallet)?; + } + // Another writer may have moved the tip while the sweep held no lock, and a + // finished key may have queued a rescan below it. + lock.write("one_time_sweep.release", || { + let scanned = wallet.block_fully_scanned()?.map(|b| b.block_height()); + if at_tip(wallet, through)? && scanned == Some(through.height) { + wallet.finish_one_time_nullifier_recovery(account, through)?; + } + Ok(()) + }) + .map_err(Error::Wallet)?; + } + if let (true, Some(transport)) = (current, transport) { + // Closed keys with no known payment need each newer publication's paid filter. + let closed = wallet.one_time_keys_to_recheck(account, through.height + 1); + if !closed.map_err(Error::Wallet)?.is_empty() { + let accepted = Directory::accept(wallet, params, through, genesis, origin, transport); + directory = Some(accepted.await?); + } + } + if let Some(directory) = &mut directory { + // Only keys this publication has not checked, so the filters are fetched only for + // a newer one. + let keys = wallet + .one_time_keys_to_recheck(account, directory.anchor.height) + .map_err(Error::Wallet)?; + if !keys.is_empty() { + let checks = directory.check(&keys, now).await?; + let checked: Vec<_> = keys + .iter() + .zip(checks) + .map(|(item, check)| (item.key, check.paid)) + .collect(); + lock.write("one_time_sweep.recheck", || { + wallet.record_paid_filter_check(account, directory.anchor, &checked) + }) + .map_err(Error::Wallet)?; + } + } + // Read last, since a paid filter hit queues its key's sweep again. + let pending = wallet + .one_time_history_pending(account, through.height) + .map_err(Error::Wallet)?; + Ok(Swept { + finished, + deferred, + pending, + }) +} + +/// Whether `through` is still the wallet's stored chain tip and the block it scanned +/// there. +fn at_tip(wallet: &W, through: ChainPoint) -> Result { + Ok(wallet.chain_height()? == Some(through.height) + && wallet.get_block_hash(through.height)? == Some(through.hash)) +} + +/// What a publication's filters say about one key's receiver. +#[derive(Clone, Copy)] +struct Check { + /// The paid set holds it, so a lookup may find payments. + paid: bool, + /// A swap may still pay it, so its key keeps scanning after its sweep. + recent: bool, + /// A swap provider's seen set holds it, so it is never issued again. + seen: bool, +} + +/// A receiver-directory publication accepted at a block the wallet scanned. Its filters +/// are fetched only once a key needs them, and its PIR session and common witnesses only +/// once a key needs a lookup or has payments to apply. +struct Directory<'a, T> { + origin: &'a str, + /// Moves into the session when one opens. + transport: Option, + manifest: receiver_pir::Manifest, + accepted: AcceptedCoverage, + anchor: ChainPoint, + filters: Option, + /// The labels of the provider seen sets. + seen: Vec, + session: Option>, +} + +/// An open PIR session and the publication's common witnesses, which every wallet +/// fetches identically. +struct Session { + client: DirectoryClient, + witnesses: WitnessSnapshot, +} + +impl<'a, T: Transport> Directory<'a, T> { + /// Accepts the publication `origin` serves, at the block of the wallet's chain it + /// ends at. + async fn accept( + wallet: &W, + params: &P, + through: ChainPoint, + genesis: [u8; 32], + origin: &'a str, + transport: T, + ) -> Result>> { + let manifest = DirectoryClient::fetch_manifest(origin, &transport).await?; + let end = BlockHeight::from(manifest.directory.end_height); + let anchor = wallet + .directory_publication_anchor(end, through) + .map_err(Error::Wallet)? + .map_err(Error::Deferred)?; + let activation = params + .activation_height(NetworkUpgrade::Nu6_3) + .ok_or(Error::Unavailable("Ironwood is not active on this network"))?; + let accepted = AcceptedCoverage { + genesis, + required_start: activation.into(), + height: anchor.height.into(), + hash: anchor.hash.0, + }; + accepted.check(&manifest.directory)?; + // Height and hash alone accept a publication that stops partway through the + // anchor block, whose omitted payments no lookup or scan from `anchor + 1` finds. + let tree_size = wallet + .block_metadata(anchor.height) + .map_err(Error::Wallet)? + .and_then(|block| block.ironwood_tree_size()); + if tree_size.map(u64::from) != Some(manifest.directory.end_position) { + return Err(Error::Unavailable( + "the receiver directory's tree size differs from the wallet's at its anchor", + )); + } + // Sweeps use only seen-set membership; issuance checks the sets' dates as it issues. + let seen = seen_sets(&manifest.directory.filters)? + .into_iter() + .map(|(label, _, _)| label.to_owned()) + .collect(); + Ok(Self { + origin, + transport: Some(transport), + manifest, + accepted, + anchor, + filters: None, + seen, + session: None, + }) + } + + /// What the filters say about each item's receiver at `now`, downloading them first + /// if needed; see [`sweep`] for how the recent sets are read. + async fn check(&mut self, work: &[DiscoveryWork], now: i64) -> Result, Error> { + let receivers = work + .iter() + .map(|item| Receiver::from_bytes(item.receiver)) + .collect::, _>>() + .map_err(DirectoryError::from)?; + if self.filters.is_none() { + let transport = self + .transport + .as_ref() + .ok_or(Error::Unavailable("the receiver directory session failed"))?; + let filters = + DirectoryClient::fetch_filters(self.origin, transport, &self.manifest).await?; + self.filters = Some(filters); + } + let key = &self.manifest.directory.salt; + let filters = self.filters.as_ref().expect("fetched above"); + // Fetching the filters checked them against the manifest's sets. + let matches = |label: &str| { + filters + .get(label) + .expect("the manifest declares every set") + .matches(key, &receivers) + }; + let mut checks: Vec<_> = matches(PAID) + .into_iter() + .map(|paid| Check { + paid, + recent: false, + seen: false, + }) + .collect(); + // Every provider has a seen set (see `seen_sets`), and needs a current, complete + // recent set too, or any receiver may still be paid. + let mut complete = true; + for label in &self.seen { + for (check, hit) in checks.iter_mut().zip(matches(label)) { + check.seen |= hit; + } + let provider = label.split_once('/').map_or(label.as_str(), |(p, _)| p); + let recent = format!("{provider}/{RECENT}"); + let sets = &self.manifest.directory.filters; + match sets.iter().find(|set| set.label == recent) { + Some(set) if covers_restore_watch(set, now) => { + for (check, hit) in checks.iter_mut().zip(matches(&recent)) { + check.recent |= hit; + } + } + _ => complete = false, + } + } + if !complete { + for check in &mut checks { + check.recent = true; + } + } + Ok(checks) + } + + /// The PIR session, opened for `lookups` lookups if none is open yet. + async fn session(&mut self, lookups: usize) -> Result<&mut Session, Error> { + if self.session.is_none() { + let transport = self + .transport + .take() + .ok_or(Error::Unavailable("the receiver directory session failed"))?; + let client = DirectoryClient::connect_manifest( + self.origin, + transport, + self.accepted, + self.manifest.clone(), + lookups, + ) + .await?; + let witnesses = client.witnesses().await?; + self.session = Some(Session { client, witnesses }); + } + Ok(self.session.as_mut().expect("opened above")) + } +} + +/// Whether a provider's recent set can rule a receiver out at `now` (see [`sweep`]). +fn covers_restore_watch(set: &FilterSet, now: i64) -> bool { + let window = set.window_secs.and_then(|w| i64::try_from(w).ok()); + match (window, set.since_unix, set.until_unix) { + (Some(window), Some(since), Some(until)) => { + window >= RESTORE_WATCH_SECS + && since <= until.saturating_sub(window) + && until >= now.saturating_sub(RECENT_SET_MAX_AGE_SECS) + } + _ => false, + } +} + +/// Sweeps one key whose attempt is leased: looks up its receiver if `check` shows the +/// paid set holds it, queues the answer, empty on a miss, with new payments' note data +/// in batches, and applies it with the publication's witnesses. `lookups` is the batch's +/// lookup count, which sizes a session that this key opens. +#[allow(clippy::too_many_arguments)] +async fn sweep_key( + wallet: &mut W, + directory: &mut Directory<'_, T>, + account: AccountId, + item: &DiscoveryWork, + check: Check, + lookups: usize, + through: ChainPoint, + notes: &mut N, + lock: &L, +) -> Result<(), Error>> +where + W: OneTimeIvkRestoreWrite, + T: Transport, + N: NoteSource, + L: WriteLock, +{ + let key = item.key; + let anchor = directory.anchor; + let paid = check.paid; + let payments: Vec<_> = if paid { + let receiver = Receiver::from_bytes(item.receiver).map_err(DirectoryError::from)?; + let accepted = directory.accepted; + let session = directory.session(lookups).await?; + session + .client + .lookup(receiver, accepted) + .await? + .into_iter() + .map(directory_payment) + .collect() + } else { + Vec::new() + }; + loop { + let batch: Vec<_> = wallet + .directory_note_data_needed(account, key, &payments) + .map_err(Error::Wallet)? + .into_iter() + .take(NOTE_BATCH) + .collect(); + let note_data = if batch.is_empty() { + BTreeMap::new() + } else { + notes.note_data(wallet, batch.clone()).await? + }; + let done = lock + .write("one_time_sweep.queue", || { + wallet.queue_directory_lookup(account, key, anchor, &payments, ¬e_data) + }) + .map_err(Error::Wallet)? + .map_err(Error::Deferred)?; + if done { + break; + } + // A requested position left without data would be asked for again forever. + if batch.iter().any(|p| !note_data.contains_key(p)) { + return Err(Error::Unavailable( + "note data is missing for a directory payment", + )); + } + } + // A miss queued nothing to apply. + let witnesses = if paid { + Some(&directory.session(lookups).await?.witnesses) + } else { + None + }; + let applied = lock + .write("one_time_sweep.apply", || { + wallet.apply_one_time_sweep(account, key, through, anchor, check.recent, |p, cmx| { + witnesses.and_then(|w| w.path(p, cmx).ok()) + }) + }) + .map_err(Error::Wallet)? + .map_err(Error::Deferred)?; + match applied { + PaymentApplication::Applied => Ok(()), + waiting => Err(Error::Queued(waiting)), + } +} + +/// A directory payment as the wallet takes it. +fn directory_payment(payment: receiver_directory::Payment) -> DirectoryPayment { + DirectoryPayment { + height: payment.height, + block_hash: payment.block_hash, + txid: payment.txid, + tx_index: payment.tx_index, + action_index: payment.action_index, + position: payment.position, + action_nullifier: payment.action_nullifier, + cmx: payment.cmx, + ephemeral_key: payment.ephemeral_key, + ciphertext_prefix: payment.ciphertext_prefix, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A recent set of `window_secs` from a feed that started at `since_unix` and last + /// read the provider at `until_unix`. + fn recent(window_secs: Option, since_unix: Option, until_unix: i64) -> FilterSet { + FilterSet { + label: "near-intents/recent".into(), + count: 0, + window_secs, + since_unix, + until_unix: Some(until_unix), + } + } + + #[test] + fn a_recent_set_rules_receivers_out_only_when_current_across_the_whole_watch() { + let now = 10 * RESTORE_WATCH_SECS; + let watch = RESTORE_WATCH_SECS as u64; + let read = now - 60; + let started = Some(read - RESTORE_WATCH_SECS); + assert!(covers_restore_watch( + &recent(Some(watch), started, read), + now + )); + assert!(covers_restore_watch( + &recent(Some(2 * watch), Some(0), read), + now + )); + // A shorter window, or a feed younger than the window, can miss a quote. + assert!(!covers_restore_watch( + &recent(Some(watch - 1), started, read), + now + )); + assert!(!covers_restore_watch( + &recent(Some(watch), Some(read - 60), read), + now + )); + assert!(!covers_restore_watch(&recent(None, started, read), now)); + assert!(!covers_restore_watch(&recent(Some(watch), None, read), now)); + // So can a set whose feed stalled: quotes after its last read are missing. + let stalled = now - RECENT_SET_MAX_AGE_SECS - 1; + let started = Some(stalled - RESTORE_WATCH_SECS); + assert!(!covers_restore_watch( + &recent(Some(watch), started, stalled), + now + )); + let mut undated = recent(Some(watch), started, read); + undated.until_unix = None; + assert!(!covers_restore_watch(&undated, now)); + } +} diff --git a/zakura/pir-receiver/tests/sweep.rs b/zakura/pir-receiver/tests/sweep.rs new file mode 100644 index 00000000..9295c7aa --- /dev/null +++ b/zakura/pir-receiver/tests/sweep.rs @@ -0,0 +1,1305 @@ +//! Restore sweeps end to end: a restored wallet looks up its one-time keys in a receiver +//! directory served in process, over the service's own routes and real PIR, and +//! imports what it finds only after checking it against its own chain. + +use std::{ + cell::{Cell, RefCell}, + collections::BTreeMap, + num::NonZeroU32, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, +}; + +use axum::{Router, body::Body, http::Request}; +use base64::Engine as _; +use orchard::{ + keys::{FullViewingKey, Scope}, + note_encryption::{CompactAction, IronwoodDomain, IronwoodNoteEncryption}, +}; +use receiver_directory::{ + Payment, Receiver, Record, + snapshot::{Manifest, PROFILE, ProviderSet, Snapshot}, + witness::WitnessSnapshot, +}; +use receiver_pir::{MIN_ROWS, server::Server}; +use receiver_pir_server::{Publication, Publications, router_with_publications}; +use rusqlite::hooks::{AuthAction, AuthContext, Authorization}; +use sha2::{Digest as _, Sha256}; +use tower::ServiceExt as _; +use zakura_one_time_ivk::swap::RESTORE_WATCH_SECS; +use zakura_pir_enhance::{ + ClientError, QueryBinding, ShardSession, + transport::{Request as EnhanceRequest, ResponseBody, Transport as EnhanceTransport}, + types, +}; +use zakura_pir_receiver::{ + DirectoryError, EnhanceNotes, Error, NoteSource, Swept, Transport, WriteLock, fetch_seen, sweep, +}; +use zcash_client_backend::data_api::{ + Account as _, WalletRead as _, WalletWrite as _, + chain::BlockSource as _, + one_time_ivk::{ + OneTimeIvkRead as _, + restore::{OneTimeIvkRestoreRead, OneTimeIvkRestoreWrite as _, PaymentApplication}, + }, + testing::{AddressType, IronwoodFvk, TestBuilder, TestRng, TestState}, + transparent_ledger::ChainPoint, +}; +use zcash_client_sqlite::{ + AccountUuid, WalletDb, + error::SqliteClientError, + testing::{ + BlockCache, + db::{TestDb, TestDbFactory}, + }, + util::{SystemClock, testing::FixedClock}, + wallet::one_time_ivk::{ + KeyId, Purpose, + swap::{ProviderSeen, RECEIVE_GAP_LIMIT}, + }, +}; +use zcash_note_encryption::{Domain as _, try_compact_note_decryption}; +use zcash_primitives::block::BlockHash; +use zcash_protocol::{ + consensus::{BlockHeight, NetworkType, NetworkUpgrade, Parameters}, + local_consensus::LocalNetwork, + value::Zatoshis, +}; + +#[path = "../../pir-enhance/src/test_support.rs"] +mod test_support; + +/// Asserts that a value matches a pattern, showing the value otherwise. +macro_rules! assert_matches { + ($value:expr, $($pattern:tt)+) => { + match &$value { + value => assert!(matches!(value, $($pattern)+), "{value:?}"), + } + }; +} + +type State = TestState; +/// The wallet a test sweeps. +type Wallet = WalletDb; + +/// The genesis hash the fixture's publications commit to. +const GENESIS: [u8; 32] = [9; 32]; +/// The origin requests name. The transport serves them in process. +const ORIGIN: &str = "https://receiver.test"; +/// Ironwood activation, the history every publication must cover. +const ACTIVATION: u32 = 100_000; +/// The caller's clock. +const NOW: i64 = 1_700_000_000; +/// The payout's value. +const VALUE: u64 = 100_000; + +/// A wallet with Ironwood active from its first block, stored as `factory` makes it. Its +/// incoming lookahead is registered before it scans, as a restore does, so scanning keeps +/// the spend evidence sweeps need. +fn ironwood_wallet_in(factory: TestDbFactory) -> State { + let activation = BlockHeight::from_u32(ACTIVATION); + let mut st = TestBuilder::new() + .with_network(LocalNetwork { + nu6: Some(activation), + nu6_1: Some(activation), + nu6_2: Some(activation), + nu6_3: Some(activation), + ..TestBuilder::<(), ()>::DEFAULT_NETWORK + }) + .with_data_store_factory(factory) + .with_block_cache(BlockCache::new()) + .with_account_from_sapling_activation(BlockHash([0; 32])) + .build(); + let account = st.test_account().unwrap().id(); + restore(&mut st, account); + st +} + +/// The wallet's fully scanned tip. +fn tip(st: &State) -> ChainPoint { + let block = st.wallet().block_fully_scanned().unwrap().unwrap(); + ChainPoint { + height: block.block_height(), + hash: block.block_hash(), + } +} + +/// The receiver of the test account's incoming one-time key `index`. +fn receiver(st: &State, index: u64) -> Receiver { + let parent = FullViewingKey::from(st.test_account().unwrap().usk().orchard()); + let fvk = KeyId::new(Purpose::Receive, index).derive(&parent).unwrap(); + Receiver::from_bytes(fvk.address_at(0u32, Scope::External).to_raw_address_bytes()).unwrap() +} + +/// Pays incoming one-time key `index` in a new scanned block. Scanning cannot find it: a +/// restored key is swept before it is ever scanned. Returns the payment as the +/// directory lists it and its note data. The output is the chain's Ironwood leaf at +/// `position`, which is the number of earlier Ironwood outputs. +fn pay_restored_key(st: &mut State, index: u64, position: u64) -> (Record, [u8; 528]) { + let parent = FullViewingKey::from(st.test_account().unwrap().usk().orchard()); + let fvk = KeyId::new(Purpose::Receive, index).derive(&parent).unwrap(); + let (height, _, _) = st.generate_next_block( + &IronwoodFvk(fvk.clone()), + AddressType::DefaultExternal, + Zatoshis::const_from_u64(VALUE), + ); + st.scan_cached_blocks_with_one_time_ivks(height, 1); + let mut block = None; + st.cache() + .with_blocks::<_, ()>(Some(height), Some(1), |cb| { + block = Some(cb); + Ok(()) + }) + .unwrap(); + let block = block.unwrap(); + let tx = &block.vtx[0]; + let compact = &tx.ironwood_actions[0]; + let action = CompactAction::try_from(compact).unwrap(); + let (note, _) = try_compact_note_decryption( + &IronwoodDomain::for_compact_action(&action), + &fvk.to_ivk(Scope::External).prepare(), + &action, + ) + .unwrap(); + // Encrypting the note again reproduces the output, with a known memo. + let encryption = IronwoodNoteEncryption::new(None, note, [4; 512]); + let ciphertext = encryption.encrypt_note_plaintext(); + assert_eq!(&ciphertext[..52], compact.ciphertext.as_slice()); + let record = Record { + receiver: receiver(st, index), + page: 0, + total: 1, + payment: Payment { + height: u32::from(height), + block_hash: block.hash().0, + txid: *tx.txid().as_ref(), + // The mock block has no coinbase; the directory refuses index 0 as one. + tx_index: u32::try_from(tx.index).unwrap() + 1, + action_index: 0, + position, + action_nullifier: compact.nullifier.as_slice().try_into().unwrap(), + cmx: compact.cmx.as_slice().try_into().unwrap(), + ephemeral_key: IronwoodDomain::epk_bytes(encryption.epk()).0, + ciphertext_prefix: ciphertext[..52].try_into().unwrap(), + }, + }; + (record, ciphertext[52..].try_into().unwrap()) +} + +/// A restored wallet at its fully scanned tip, after payments to its incoming keys. +struct Restored { + st: State, + account: AccountUuid, + through: ChainPoint, + /// The payments as the directory lists them, in chain order. + records: Vec, + /// Their note data, by position. + data: BTreeMap, +} + +/// Pays incoming key `paid[i]` as the chain's Ironwood output `i`, each payment a page of +/// its key's lookup, then scans one more block and registers the restored lookahead. +fn restored(paid: &[u64]) -> Restored { + let mut st = ironwood_wallet_in(TestDbFactory::default()); + let account = st.test_account().unwrap().id(); + let (mut records, mut data) = (Vec::new(), BTreeMap::new()); + for (position, index) in (0..).zip(paid) { + let (mut record, note_data) = pay_restored_key(&mut st, *index, position); + let pages = |paid: &[u64]| paid.iter().filter(|i| *i == index).count(); + record.page = pages(&paid[..records.len()]).try_into().unwrap(); + record.total = pages(paid).try_into().unwrap(); + records.push(record); + data.insert(position, note_data); + } + st.generate_and_scan_empty_blocks_with_one_time_ivks(1); + restore(&mut st, account); + let through = tip(&st); + Restored { + st, + account, + through, + records, + data, + } +} + +impl Restored { + /// A note source serving every payment's note data. + fn notes(&self) -> Notes { + let data = self.data.clone(); + Notes { + data, + ..Notes::default() + } + } +} + +/// The test account's total balance in zatoshis. +fn balance(st: &State) -> u64 { + st.get_total_balance(st.test_account().unwrap().id()) + .into_u64() +} + +/// Whether the test account still has sweeps or queued payments through `through`. +fn pending(st: &State, through: ChainPoint) -> bool { + let account = st.test_account().unwrap().id(); + let db = st.wallet().db(); + db.one_time_history_pending(account, through.height) + .unwrap() +} + +/// The indices of the keys the wallet scans with, in order. +fn scanning(st: &State) -> Vec { + let keys = st.wallet().get_one_time_scanning_keys().unwrap(); + let mut indices: Vec<_> = keys.iter().map(|k| k.key_id().index()).collect(); + indices.sort(); + indices +} + +/// A swap provider's sets: an empty recent set covering the wallet's restore watch, and +/// every receiver it ever had, from a feed that started long before [`NOW`] and last +/// read the provider a minute before it. +fn provider(seen: &[Receiver]) -> Vec { + provider_named("near-intents", seen) +} + +/// [`provider`]'s sets for the provider `name`. +fn provider_named(name: &str, seen: &[Receiver]) -> Vec { + let since_unix = NOW - 2 * RESTORE_WATCH_SECS; + let until_unix = NOW - 60; + vec![ + ProviderSet { + label: format!("{name}/recent"), + window_secs: Some(RESTORE_WATCH_SECS.try_into().unwrap()), + since_unix, + until_unix, + receivers: Vec::new(), + }, + ProviderSet { + label: format!("{name}/seen"), + window_secs: None, + since_unix, + until_unix, + receivers: seen.to_vec(), + }, + ] +} + +/// A publication of `records` ending at `end`, with the common witnesses over the +/// chain's Ironwood `commitments`, if any, and the swap `provider` sets in its filters, +/// served by the receiver service's routes. +fn publish( + records: &[Record], + commitments: &[[u8; 32]], + end: ChainPoint, + provider: &[ProviderSet], +) -> Router { + let manifest = Manifest { + profile: PROFILE.into(), + genesis: GENESIS, + start_height: ACTIVATION, + start_parent: [0; 32], + start_position: 0, + end_height: end.height.into(), + end_hash: end.hash.0, + end_position: commitments.len() as u64, + rows: MIN_ROWS, + salt: [5; 32], + records: 0, + data_sha256: [0; 32], + filters: Vec::new(), + filters_sha256: [0; 32], + }; + let snapshot = Snapshot::build(manifest, records, provider).unwrap(); + let positions = records + .iter() + .map(|record| u32::try_from(record.payment.position).unwrap()) + .collect(); + // A chain without Ironwood outputs has no witnesses to serve. + let witnesses = (!commitments.is_empty()).then(|| { + WitnessSnapshot::build(&snapshot.manifest, commitments, &positions) + .unwrap() + .encode() + }); + let publication = Publication::new(Server::new(snapshot).unwrap(), witnesses).unwrap(); + let publications = Publications::default(); + assert!(publications.publish(publication, publications.epoch())); + router_with_publications(publications) +} + +/// Serves a [`publish`]ed publication of `records`, the chain's only Ironwood outputs, +/// ending at `end`, with a provider that has seen nothing. +fn serve(records: &[Record], end: ChainPoint) -> Directory { + let commitments: Vec<_> = records.iter().map(|record| record.payment.cmx).collect(); + Directory::new(publish(records, &commitments, end, &provider(&[]))) +} + +/// Serves a directory's routes in process and records every request. +struct Directory { + router: Router, + requests: Mutex>, +} + +impl Directory { + /// Serves `router` at [`ORIGIN`]. + fn new(router: Router) -> Self { + Self { + router, + requests: Mutex::new(Vec::new()), + } + } + + /// Every request so far, as method and path. + fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + /// Sends one request to the routes, failing over `limit` bytes as a transport must. + async fn send( + &self, + method: &str, + url: &str, + body: Vec, + limit: usize, + ) -> Result, DirectoryError> { + let path = url + .strip_prefix(ORIGIN) + .expect("requests stay on the origin"); + self.requests + .lock() + .unwrap() + .push(format!("{method} {path}")); + let request = Request::builder() + .method(method) + .uri(path) + .body(Body::from(body)) + .unwrap(); + let response = self.router.clone().oneshot(request).await.unwrap(); + let status = response.status().as_u16(); + let body = axum::body::to_bytes(response.into_body(), limit) + .await + .map_err(|_| DirectoryError::Transport("response over its limit".into()))?; + match status { + 200 => Ok(body.to_vec()), + 409 | 410 => Err(DirectoryError::Revision), + status => Err(DirectoryError::Transport(format!("HTTP {status}"))), + } + } +} + +impl Transport for Directory { + async fn get(&self, url: &str, limit: usize) -> Result, DirectoryError> { + self.send("GET", url, Vec::new(), limit).await + } + + async fn post( + &self, + url: &str, + body: Vec, + limit: usize, + ) -> Result, DirectoryError> { + self.send("POST", url, body, limit).await + } +} + +/// The receiver lookups `directory` has served. +fn lookups(directory: &Directory) -> usize { + let requests = directory.requests(); + requests + .iter() + .filter(|r| *r == "POST /v1/receiver/query") + .count() +} + +/// Note data from a fixed map, recording each request. Each answer is keyed `shift` +/// positions after its request, and an `unavailable` source cannot serve the wallet's +/// chain. +#[derive(Default)] +struct Notes { + data: BTreeMap, + requested: Vec>, + shift: u64, + unavailable: bool, +} + +impl Notes { + /// A source that cannot serve the wallet's chain. + fn unavailable() -> Self { + let unavailable = true; + Self { + unavailable, + ..Self::default() + } + } +} + +impl NoteSource for Notes { + async fn note_data( + &mut self, + _: &W, + positions: Vec, + ) -> Result, Error> { + if self.unavailable { + return Err(Error::NotesUnavailable("no note data")); + } + self.requested.push(positions.clone()); + assert!( + self.requested.len() <= 3, + "the sweep keeps asking for {positions:?}" + ); + let answer = |p| Some((p + self.shift, *self.data.get(&p)?)); + Ok(positions.into_iter().filter_map(answer).collect()) + } +} + +/// A test needs no lock: nothing else writes to its wallet. +struct NoLock; + +impl WriteLock for NoLock { + fn write(&self, _: &'static str, write: impl FnOnce() -> T) -> T { + write() + } +} + +/// Records each write's label. +#[derive(Default)] +struct Labels(RefCell>); + +impl WriteLock for Labels { + fn write(&self, label: &'static str, write: impl FnOnce() -> T) -> T { + self.0.borrow_mut().push(label); + write() + } +} + +/// Registers the restored incoming lookahead, as a restore does at the tip. +fn restore(st: &mut State, account: AccountUuid) { + st.wallet_mut() + .db_mut() + .maintain_one_time_ivks(account) + .unwrap(); +} + +/// Scans every range the wallet has queued, such as a spend-history replay. +fn scan_queued(st: &mut State) { + for range in st.wallet().suggest_scan_ranges().unwrap() { + let blocks = range.block_range(); + let count = u32::from(blocks.end) - u32::from(blocks.start); + st.scan_cached_blocks_with_one_time_ivks(blocks.start, count as usize); + } +} + +/// Runs one sweep of the test account through `through` at `now`. +async fn run>( + st: &mut State, + through: ChainPoint, + directory: impl Transport, + notes: &mut N, + lock: &impl WriteLock, + now: i64, +) -> Result, Error> { + let network = *st.network(); + let account = st.test_account().unwrap().id(); + sweep( + st.wallet_mut().db_mut(), + &network, + account, + through, + GENESIS, + ORIGIN, + directory, + notes, + lock, + now, + ) + .await +} + +/// [`run`]s without a lock, expecting no key deferred. +async fn run_clean>( + st: &mut State, + through: ChainPoint, + directory: impl Transport, + notes: &mut N, + now: i64, +) -> Swept { + let swept = run(st, through, directory, notes, &NoLock, now) + .await + .unwrap(); + assert!(swept.deferred.is_empty(), "{:?}", swept.deferred); + swept +} + +/// Runs sweeps at [`NOW`] until none is pending, expecting no deferral, and returns how +/// many keys each run finished. +async fn settle( + st: &mut State, + through: ChainPoint, + directory: &Directory, + notes: &mut Notes, +) -> Vec { + let mut finished = Vec::new(); + for _ in 0..20 { + let swept = run_clean(st, through, directory, notes, NOW).await; + finished.push(swept.finished); + if !swept.pending { + return finished; + } + } + panic!("sweeps are still pending after 20 runs"); +} + +/// A restore sweep imports every payment to a key, each a page of its lookup, with note +/// data in batches of 64, then moves the lookahead past the paid and `seen` indices, one +/// batch per run, so later runs find a payment far past the first batch. Every request +/// stays on the service's routes: a wallet without payments downloads only a manifest and +/// the filters, and a few pages cost less over PIR than the row file, unlike many more. +#[tokio::test] +async fn a_restore_sweep_imports_every_payment_and_moves_the_lookahead_past_it() { + for (paid, seen, batches, row_file) in [ + (vec![], 0, vec![], false), + (vec![3], 0, vec![1], false), + (vec![1; 2], 0, vec![2], false), + (vec![1; 65], 0, vec![64, 1], true), + // The set is finite, so an unbounded run would stop past it instead of hanging. + (vec![100], 200, vec![1], false), + ] { + let mut r = restored(&paid); + assert_eq!(balance(&r.st), 0); + let seen: Vec<_> = (0..seen).map(|i| receiver(&r.st, i)).collect(); + let commitments: Vec<_> = r.records.iter().map(|r| r.payment.cmx).collect(); + let sets = provider(&seen); + let directory = Directory::new(publish(&r.records, &commitments, r.through, &sets)); + let mut notes = r.notes(); + let runs = settle(&mut r.st, r.through, &directory, &mut notes).await; + // Each run sweeps at most one batch, and later runs progress. + assert!(runs.iter().all(|f| (1..=64).contains(f)), "{runs:?}"); + let finished: usize = runs.iter().sum(); + // The first window, then, in later runs, the keys above it that the paid and seen + // indices bring in. + let past = paid + .last() + .map_or(0, |index| index + 1) + .max(seen.len() as u64); + assert_eq!(finished as u64, RECEIVE_GAP_LIMIT + past, "{paid:?}"); + let sizes: Vec<_> = notes.requested.iter().map(Vec::len).collect(); + assert_eq!(sizes, batches); + let positions: Vec<_> = (0..paid.len() as u64).collect(); + assert_eq!(notes.requested.concat(), positions); + assert_eq!(balance(&r.st), paid.len() as u64 * VALUE); + let requests = directory.requests(); + let routes = + |r: &String| r.starts_with("GET /v1/receiver/") || r == "POST /v1/receiver/query"; + assert!(requests.iter().all(routes), "{requests:?}"); + let rows = |r: &String| r.starts_with("GET /v1/receiver/rows/"); + assert_eq!(requests.iter().any(rows), row_file, "{requests:?}"); + if paid.is_empty() { + // No PIR session, witnesses or note data, and no second run. + assert_eq!(requests.len(), 2, "{requests:?}"); + assert!(requests[1].starts_with("GET /v1/receiver/filters/")); + } + } +} + +/// A key closed after its sweep is checked against each newer publication's paid filter, +/// so a payment after the first publication's anchor is found once a publication shows +/// it, and not before. A publication already checked costs only its manifest, and a key +/// with a known payment is not swept again. +#[tokio::test] +async fn a_newer_paid_filter_finds_a_payment_to_a_closed_key() { + let mut r = restored(&[]); + let directory = serve(&[], r.through); + let mut notes = r.notes(); + settle(&mut r.st, r.through, &directory, &mut notes).await; + assert!(scanning(&r.st).is_empty()); + + // The closed key misses a payment, which the same publication cannot show. + let (record, note_data) = pay_restored_key(&mut r.st, 0, 0); + let through = tip(&r.st); + let before = directory.requests().len(); + let swept = run(&mut r.st, through, &directory, &mut notes, &NoLock, NOW).await; + assert!(!swept.unwrap().pending); + assert_eq!(directory.requests()[before..], ["GET /v1/receiver/init"]); + assert_eq!(balance(&r.st), 0); + + let directory = serve(std::slice::from_ref(&record), through); + notes.data.insert(record.payment.position, note_data); + // The paid key, then the lookahead key its index brings in. + let runs = settle(&mut r.st, through, &directory, &mut notes).await; + assert_eq!(runs.iter().sum::(), 2); + assert_eq!(balance(&r.st), VALUE); + + // A later publication still holds the paid key, but it is not looked up again. + r.st.generate_and_scan_empty_blocks_with_one_time_ivks(1); + let later = tip(&r.st); + let directory = serve(std::slice::from_ref(&record), later); + let swept = run(&mut r.st, later, &directory, &mut notes, &NoLock, NOW).await; + let swept = swept.unwrap(); + assert_eq!((swept.finished, swept.pending), (0, false)); + assert!(directory.requests().iter().all(|r| !r.starts_with("POST"))); +} + +/// After its sweep, a key keeps scanning only if a provider's recent set holds its +/// receiver, and every key does unless each provider has a recent set. +#[tokio::test] +async fn only_recently_quoted_addresses_keep_scanning_after_their_sweep() { + for (case, expected) in [ + ("quoted", vec![2]), + ("both recent", vec![]), + ("one recent", (0..RECEIVE_GAP_LIMIT).collect()), + ] { + let mut r = restored(&[]); + let sets = if case == "quoted" { + // The old device quoted addresses 0 and 2, and 2 only today. + let mut sets = provider(&[receiver(&r.st, 0), receiver(&r.st, 2)]); + sets[0].receivers = vec![receiver(&r.st, 2)]; + sets + } else { + // Provider `b` publishes only its seen set in the last case. + let mut sets = provider_named("a", &[]); + let b = provider_named("b", &[]).into_iter(); + sets.extend(b.skip(usize::from(case == "one recent"))); + sets + }; + let directory = Directory::new(publish(&[], &[], r.through, &sets)); + let mut notes = r.notes(); + settle(&mut r.st, r.through, &directory, &mut notes).await; + assert_eq!(scanning(&r.st), expected, "{case}"); + } +} + +/// A seen hit counts from the attempt that observes it. When publication A's seen set +/// holds a receiver whose payment it then misstates, a later publication whose seen set +/// misses the receiver and whose lookup finds nothing does not make it issuable. +#[tokio::test] +async fn a_seen_hit_outlives_a_rejected_payment_and_a_later_miss() { + // Key 1000's later output is one that publication A's tree replaces. + let mut r = restored(&[0, 1000]); + let key = KeyId::new(Purpose::Receive, 0); + let record = &r.records[0]; + let commitments = [record.payment.cmx, r.records[1].payment.cmx]; + let seen = |st: &State| -> bool { + st.wallet() + .conn() + .query_row( + "SELECT provider_seen FROM ironwood_receiving_keys + WHERE purpose = 1 AND key_index = ?1", + [key.index().to_be_bytes()], + |r| r.get(0), + ) + .unwrap() + }; + let mut notes = r.notes(); + // Publication A's witnesses are over another tree, so its payment is rejected. + let a = Directory::new(publish( + std::slice::from_ref(record), + &[record.payment.cmx, [1; 32]], + r.through, + &provider(&[record.receiver]), + )); + let swept = run(&mut r.st, r.through, &a, &mut notes, &NoLock, NOW) + .await + .unwrap(); + let rejected = PaymentApplication::Rejected; + assert_matches!(swept.deferred[..], [(k, Error::Queued(a))] if *k == key && *a == rejected); + assert!(seen(&r.st)); + + let b = Directory::new(publish(&[], &commitments, r.through, &provider(&[]))); + let later = NOW + 3_600; + let swept = run_clean(&mut r.st, r.through, &b, &mut notes, later).await; + assert!(!swept.pending); + assert!(seen(&r.st)); + let fetched = fetch_seen(ORIGIN, &b, GENESIS).await.unwrap(); + let contains = |receivers: &[[u8; 43]]| fetched.contains(receivers); + let view = ProviderSeen { + since: fetched.since(), + until: fetched.until(), + contains: &contains, + }; + let issued = + r.st.wallet_mut() + .db_mut() + .prepare_receive_reservation(r.account, later, r.through.height, Some(&view)) + .unwrap() + .unwrap(); + assert_eq!(issued.key_id(), KeyId::new(Purpose::Receive, 1)); +} + +/// Records each write's label, and arms `fail` for the first attempt's write, which +/// the wallet's commit hook then fails. +struct FailFirstAttempt { + fail: Arc, + labels: RefCell>, +} + +impl WriteLock for FailFirstAttempt { + fn write(&self, label: &'static str, write: impl FnOnce() -> T) -> T { + let mut labels = self.labels.borrow_mut(); + if label == "one_time_sweep.attempt" && !labels.contains(&label) { + self.fail.store(true, Ordering::SeqCst); + } + labels.push(label); + drop(labels); + write() + } +} + +/// A failure to lease an attempt stops the run with the wallet's error, before that +/// key's lookup and without retrying it. +#[tokio::test] +async fn a_failed_attempt_write_stops_the_run() { + let mut r = restored(&[0]); + let directory = serve(&r.records, r.through); + let mut notes = r.notes(); + let lock = FailFirstAttempt { + fail: Arc::new(AtomicBool::new(false)), + labels: RefCell::new(Vec::new()), + }; + let fail = lock.fail.clone(); + // Returning true turns the commit into a rollback, which fails the write. + r.st.wallet() + .conn() + .commit_hook(Some(move || fail.swap(false, Ordering::SeqCst))); + let result = run(&mut r.st, r.through, &directory, &mut notes, &lock, NOW).await; + assert_matches!(result, Err(Error::Wallet(_))); + assert_eq!( + *lock.labels.borrow(), + ["one_time_sweep.due", "one_time_sweep.attempt"] + ); + assert!(directory.requests().iter().all(|r| !r.starts_with("POST"))); + assert!(notes.requested.is_empty()); +} + +/// A `through` the wallet's chain has moved past is not a tip to sweep or release +/// evidence at, so a run with it writes and requests nothing. While a block below the +/// tip is unscanned, a run prepares no sweep: it makes no request and leases no attempt, +/// since any payment it found could only wait for scanning. Once scanning catches up, +/// the sweep runs at once, without backoff. +#[tokio::test] +async fn a_run_off_the_scanned_tip_leases_and_requests_nothing() { + let mut r = restored(&[0]); + let stale = r.through; + // A block below the tip is not scanned yet. + r.st.generate_empty_block(); + let (top, _) = r.st.generate_empty_block(); + r.st.scan_cached_blocks_with_one_time_ivks(top, 1); + let block = r.st.wallet().block_metadata(top).unwrap().unwrap(); + let through = ChainPoint { + height: top, + hash: block.block_hash(), + }; + let directory = serve(&r.records, through); + let mut notes = r.notes(); + let behind = ["one_time_sweep.due", "one_time_sweep.release"]; + for (through, labels) in [(stale, &[][..]), (through, &behind[..])] { + let lock = Labels::default(); + let swept = run(&mut r.st, through, &directory, &mut notes, &lock, NOW) + .await + .unwrap(); + assert!(swept.deferred.is_empty(), "{:?}", swept.deferred); + assert!(swept.pending); + assert_eq!(*lock.0.borrow(), labels); + let requests = directory.requests(); + assert!(requests.is_empty(), "{requests:?}"); + } + r.st.scan_cached_blocks_with_one_time_ivks(top - 1, 1); + run_clean(&mut r.st, through, &directory, &mut notes, NOW).await; + settle(&mut r.st, through, &directory, &mut notes).await; + assert_eq!(balance(&r.st), VALUE); +} + +/// Another handle on the wallet's file, as another writer holds. +type OtherWriter = WalletDb; + +/// Runs `other` just before the first release, as another writer may while the sweep +/// holds no lock, then sets `releasing` while release holds the lock. +struct WriterBeforeRelease { + other: RefCell>, + releasing: Arc, +} + +impl WriteLock for WriterBeforeRelease { + fn write(&self, label: &'static str, write: impl FnOnce() -> T) -> T { + if label != "one_time_sweep.release" { + return write(); + } + if let Some(other) = self.other.borrow_mut().take() { + other(); + } + self.releasing.store(true, Ordering::SeqCst); + let result = write(); + self.releasing.store(false, Ordering::SeqCst); + result + } +} + +/// Release checks the chain point again under its lock, since another writer may move +/// the tip while the sweep's lookups run without it, and waits for the fully scanned +/// height to reach it. Each case sweeps against a publication `behind` blocks below the +/// tip, with another writer moving the chain through a second handle just before +/// release, and judges whether release ran by its read of the retention table. +#[tokio::test] +async fn release_runs_only_at_the_fully_scanned_tip() { + for (case, behind, expected) in [ + ("an unmoved tip", 0, true), + ("an advanced tip", 0, false), + ("a reorg that rewound the tip", 0, false), + // The tip is back at `through`'s height, but its block is not scanned yet. + ("a replaced block", 0, false), + // Finishing keys against a publication behind `through` rescans the blocks + // after it, lowering the fully scanned height without moving the tip. + ("a rescan the sweep queued", 1, false), + ] { + let mut st = ironwood_wallet_in(TestDbFactory::file_backed()); + let network = *st.network(); + st.generate_and_scan_empty_blocks_with_one_time_ivks(2); + let account = st.test_account().unwrap().id(); + restore(&mut st, account); + let through = tip(&st); + let path = st.wallet().conn().path().unwrap().to_owned(); + let mut handle = + OtherWriter::for_path(path, network, SystemClock, TestRng::seed_from_u64(0)).unwrap(); + let h = through.height; + let other = || match case { + "an advanced tip" => handle.update_chain_tip(h + 1).unwrap(), + "a reorg that rewound the tip" => { + handle.truncate_to_height(h - 1).unwrap(); + } + "a replaced block" => { + handle.truncate_to_height(h - 1).unwrap(); + handle.update_chain_tip(h).unwrap(); + } + _ => {} + }; + let lock = WriterBeforeRelease { + other: RefCell::new(Some(other)), + releasing: Arc::new(AtomicBool::new(false)), + }; + let released = Arc::new(AtomicBool::new(false)); + let (releasing, read) = (lock.releasing.clone(), released.clone()); + st.wallet() + .conn() + .authorizer(Some(move |ctx: AuthContext<'_>| { + let retention = matches!( + ctx.action, + AuthAction::Read { + table_name: "ironwood_one_time_spend_retention", + .. + } + ); + if releasing.load(Ordering::SeqCst) && retention { + read.store(true, Ordering::SeqCst); + } + Authorization::Allow + })); + let height = through.height - behind; + let hash = st.wallet().get_block_hash(height).unwrap().unwrap(); + // Without a recent set, every key keeps scanning from the block after the end. + let end = ChainPoint { height, hash }; + let directory = Directory::new(publish(&[], &[], end, &provider(&[])[1..])); + let swept = run( + &mut st, + through, + &directory, + &mut Notes::unavailable(), + &lock, + NOW, + ) + .await + .unwrap(); + assert_eq!(swept.finished as u64, RECEIVE_GAP_LIMIT); + st.wallet() + .conn() + .authorizer(None::) -> Authorization>); + assert_eq!(released.load(Ordering::SeqCst), expected, "{case}"); + } +} + +/// Seen sets must commit to the wallet's genesis, as a sweep's publication must: another +/// network's sets say nothing about this network's quotes. A publication without seen +/// sets, or with a provider that has a recent set but no seen set, may omit any receiver, +/// so issuance refuses it as sweeps do. Once every provider has a dated seen set, the +/// sets cover the quote times they all cover, and hold what any of them holds. +#[tokio::test] +async fn seen_sets_need_this_network_and_every_provider() { + let r = restored(&[]); + // Provider `a` has its seen set only if `a_seen`. `b` has only a seen set, from a + // later feed start and a later read. + let sets = |a_seen: bool| { + let mut sets = provider_named("a", &[receiver(&r.st, 0)]); + sets.truncate(1 + usize::from(a_seen)); + let mut b = provider_named("b", &[receiver(&r.st, 1)]).pop().unwrap(); + b.since_unix = NOW - RESTORE_WATCH_SECS; + b.until_unix = NOW - 30; + sets.push(b); + sets + }; + for sets in [sets(false), Vec::new()] { + let directory = Directory::new(publish(&[], &[], r.through, &sets)); + let refused = fetch_seen(ORIGIN, &directory, GENESIS).await; + assert!(matches!(refused, Err(DirectoryError::Malformed))); + } + + let directory = Directory::new(publish(&[], &[], r.through, &sets(true))); + let refused = fetch_seen(ORIGIN, &directory, [8; 32]).await; + let coverage = |e: &_| matches!(e, receiver_directory::Error::Coverage); + assert_matches!(refused.err(), Some(DirectoryError::Directory(e)) if coverage(e)); + let seen = fetch_seen(ORIGIN, &directory, GENESIS).await.unwrap(); + assert_eq!( + (seen.since(), seen.until()), + (NOW - RESTORE_WATCH_SECS, NOW - 60) + ); + let held = [0, 1, 2].map(|index| *receiver(&r.st, index).as_bytes()); + assert_eq!(seen.contains(&held), [true, true, false]); +} + +/// Which part of a sweep fails. +#[derive(Clone, Copy, Debug, PartialEq)] +enum Fault { + /// Queries fail after the session was accepted. + Queries, + /// The witness snapshot is malformed, so no session opens. + Witnesses, + /// The note source cannot serve the wallet's chain. + Notes, + /// Note data comes back for other positions. + OtherPositions, +} + +/// A directory with `fault`. +struct Faulty { + directory: Directory, + fault: Fault, +} + +impl Transport for Faulty { + async fn get(&self, url: &str, limit: usize) -> Result, DirectoryError> { + if matches!(self.fault, Fault::Witnesses) && url.contains("/v1/receiver/witness/") { + return Ok(vec![0]); + } + self.directory.get(url, limit).await + } + + async fn post( + &self, + url: &str, + body: Vec, + limit: usize, + ) -> Result, DirectoryError> { + if matches!(self.fault, Fault::Queries) { + return Err(DirectoryError::Revision); + } + self.directory.post(url, body, limit).await + } +} + +/// A revoked session, a session that fails to open, or a note source that cannot serve +/// the wallet's chain would fail every later key the same way, so the run stops at the +/// first paid key, and later paid keys are neither leased nor backed off. Note data for +/// other positions fails only that key, without asking for it again. +#[tokio::test] +async fn a_run_fatal_failure_stops_the_run_before_leasing_more_keys() { + for fault in [ + Fault::Queries, + Fault::Witnesses, + Fault::Notes, + Fault::OtherPositions, + ] { + // The batch's first two keys are paid, so each needs a lookup. + let mut r = restored(&[0, 1]); + let directory = Faulty { + directory: serve(&r.records, r.through), + fault, + }; + let mut notes = match fault { + Fault::Notes => Notes::unavailable(), + Fault::OtherPositions => Notes { + shift: 1, + ..r.notes() + }, + _ => r.notes(), + }; + let swept = run(&mut r.st, r.through, &directory, &mut notes, &NoLock, NOW).await; + let swept = swept.unwrap(); + let fatal = fault != Fault::OtherPositions; + let keys: Vec<_> = swept.deferred.iter().map(|(key, _)| key.index()).collect(); + assert_eq!(keys, if fatal { &[0][..] } else { &[0, 1] }, "{fault:?}"); + for (_, e) in &swept.deferred { + let failed = match fault { + Fault::Queries => matches!(e, Error::Directory(DirectoryError::Revision)), + Fault::Witnesses => matches!(e, Error::Directory(_)), + Fault::Notes => matches!(e, Error::NotesUnavailable(_)), + Fault::OtherPositions => matches!( + e, + Error::Unavailable("note data is missing for a directory payment") + ), + }; + assert!(failed, "{fault:?}: {e:?}"); + } + // Unless the run stopped, every unpaid key in the batch finishes. + let unpaid = if fatal { + 0 + } else { + RECEIVE_GAP_LIMIT as usize - 2 + }; + assert_eq!((swept.finished, swept.pending), (unpaid, true)); + assert_eq!(notes.requested.len(), keys.len() * usize::from(!fatal)); + assert_eq!(balance(&r.st), 0); + let due: Vec<_> = + r.st.wallet_mut() + .db_mut() + .prepare_one_time_sweeps(r.account, r.through, NOW, NonZeroU32::new(64).unwrap()) + .unwrap() + .unwrap() + .iter() + .map(|item| item.key.index()) + .collect(); + assert!( + !due.contains(&0) && due.contains(&1) == fatal, + "{fault:?}: {due:?}" + ); + } +} + +/// How a sweep refuses an unusable publication. +type Refusal = fn(&Error) -> bool; + +/// A publication off the wallet's chain, or ending at the wallet's block but short of +/// its last Ironwood output, whose omitted payments neither a lookup nor a rescan would +/// find, or one that may omit any seen receiver, is refused before any lookup, key finish +/// or release. Every due key backs off, so the directory is not asked again until a +/// minute on. +#[tokio::test] +async fn an_unusable_publication_is_refused_before_any_lookup() { + let malformed: Refusal = |e| matches!(e, Error::Directory(DirectoryError::Malformed)); + let rows: [(&str, Refusal); 4] = [ + ("off chain", |e| matches!(e, Error::Directory(_))), + ("short", |e| matches!(e, Error::Unavailable(_))), + ("unseen provider", malformed), + ("no seen sets", malformed), + ]; + for (case, refused) in rows { + let mut r = restored(&[0]); + let (records, through) = (&r.records[..], r.through); + let directory = match case { + "off chain" => serve( + records, + ChainPoint { + hash: BlockHash([1; 32]), + ..through + }, + ), + // Without the chain's one Ironwood output. + "short" => serve(&[], through), + // A provider with a recent set but no seen set, or no provider at all. + _ => { + let mut sets = provider(&[]); + sets.truncate(usize::from(case == "unseen provider")); + let commitments = [records[0].payment.cmx]; + Directory::new(publish(records, &commitments, through, &sets)) + } + }; + let mut notes = r.notes(); + let lock = Labels::default(); + let result = run(&mut r.st, through, &directory, &mut notes, &lock, NOW).await; + assert!(result.as_ref().is_err_and(refused), "{case}: {result:?}"); + let attempts = ["one_time_sweep.due", "one_time_sweep.attempt"]; + let labels = lock.0.take(); + assert!( + labels.iter().all(|l| attempts.contains(l)), + "{case}: {labels:?}" + ); + assert_eq!(directory.requests(), ["GET /v1/receiver/init"], "{case}"); + assert!(notes.requested.is_empty()); + assert!(pending(&r.st, through)); + for (now, requests) in [(NOW, 1), (NOW + 60, 2)] { + let _ = run(&mut r.st, through, &directory, &mut notes, &NoLock, now).await; + assert_eq!(directory.requests().len(), requests, "{case}"); + } + } +} + +/// The test network's upgrades under mainnet's network type, the only one Enhance PIR +/// accepts. +#[derive(Clone)] +struct EnhanceNetwork(LocalNetwork); + +impl Parameters for EnhanceNetwork { + fn network_type(&self) -> NetworkType { + NetworkType::Main + } + + fn activation_height(&self, nu: NetworkUpgrade) -> Option { + self.0.activation_height(nu) + } +} + +/// An Enhance service over an all-zero database whose first query is refused with +/// `status`. It counts manifest fetches and queries. +struct RefusingEnhance { + manifest: Vec, + session: Vec, + rows: u64, + status: Cell>, + inits: Cell, + queries: Cell, +} + +impl RefusingEnhance { + /// A service whose manifest ends at `anchor`, covering `records` Ironwood outputs. + fn new(anchor: ChainPoint, records: u64, status: u16) -> Self { + let public = |rows| vec![0; types::session_public_len(rows).unwrap()]; + let mut manifest = test_support::synthetic_manifest( + records, + |shard| hex::encode(Sha256::digest(public(shard.logical_rows))), + &"00".repeat(32), + ); + let rows = manifest.coverage.shards[0].logical_rows; + manifest.anchor_height = u32::from(anchor.height).into(); + manifest.anchor_block_hash = anchor.hash.to_string(); + let session = ShardSession { + session_id: hex::encode(manifest.session_id(0).unwrap()), + generation: manifest.generation, + shard_id: 0, + params: types::parameters(rows).unwrap(), + public_params_base64: base64::engine::general_purpose::STANDARD.encode(public(rows)), + }; + Self { + manifest: serde_json::to_vec(&manifest).unwrap(), + session: serde_json::to_vec(&session).unwrap(), + rows, + status: Cell::new(Some(status)), + inits: Cell::new(0), + queries: Cell::new(0), + } + } +} + +impl EnhanceTransport for RefusingEnhance { + async fn execute(&self, request: EnhanceRequest) -> Result { + let bytes = if request.url.ends_with("/v1/enhance/init") { + self.inits.set(self.inits.get() + 1); + self.manifest.clone() + } else if request.url.contains("/v1/enhance/session/") { + self.session.clone() + } else if request.url.ends_with("/v1/enhance/query") { + self.queries.set(self.queries.get() + 1); + if let Some(status) = self.status.take() { + return Err(ClientError::HttpStatus(status)); + } + // Zero public material and a zero response decode to an all-zero row. + let mut response = QueryBinding::decode(&request.body).unwrap().encode(); + response.resize(types::response_len(self.rows).unwrap(), 0); + response + } else { + panic!("unexpected endpoint: {}", request.url); + }; + let mut body = request.response_body(); + body.extend(&bytes)?; + Ok(body.finish()) + } +} + +/// Enhance notes open no session until a position needs note data, accept fresh routing +/// after a refused query, and cannot serve a wallet whose chain lacks their anchor. +#[tokio::test] +async fn enhance_notes_follow_the_wallets_chain_and_fresh_routing() { + let r = restored(&[0]); + let (anchor, db) = (r.through, r.st.wallet().db()); + let block = db.block_metadata(anchor.height).unwrap().unwrap(); + let records = block.ironwood_tree_size().unwrap().into(); + let network = EnhanceNetwork(*r.st.network()); + for status in [409, 410] { + let enhance = RefusingEnhance::new(anchor, records, status); + let mut notes = EnhanceNotes::new("https://enhance.test", &enhance, &network); + let data = notes.note_data(db, Vec::new()).await.unwrap(); + assert!(data.is_empty()); + assert_eq!(enhance.inits.get(), 0); + let refused = notes.note_data(db, vec![0]).await; + assert_matches!(refused, Err(Error::NoteData(ClientError::HttpStatus(s))) if *s == status); + // The same notes fetch and accept the routing again before the next batch. + let data = notes.note_data(db, vec![0]).await.unwrap(); + assert_eq!(data, BTreeMap::from([(0, [0; 528])])); + assert_eq!((enhance.inits.get(), enhance.queries.get()), (2, 2)); + } + // Enhance's anchor is at the wallet's tip height but on another chain. + let other = ChainPoint { + hash: BlockHash([7; 32]), + ..anchor + }; + let enhance = RefusingEnhance::new(other, records, 409); + let mut notes = EnhanceNotes::new("https://enhance.test", &enhance, &network); + let refused = notes.note_data(db, vec![0]).await; + assert_matches!(refused, Err(Error::NotesUnavailable(_))); + assert_eq!(enhance.queries.get(), 0); +} + +/// What a later publication answers for a deferred candidate. +#[derive(Clone, Copy, Debug, PartialEq)] +enum Answer { + /// The payment the first publication misstated. + Corrected, + /// The same payment. + Unchanged, + /// Nothing: its paid filter misses the receiver. + Dropped, +} + +/// A queued candidate that defers is replaced by each later attempt's answer from the +/// publication it uses, whatever made it wait: a corrected payment is recovered with +/// fresh note data, an unchanged one reuses the queued note data, and a paid-filter miss +/// drops the candidate. A deferral that queued a replay holds the next attempt, without +/// a request, until the replay is scanned. +#[tokio::test] +async fn a_deferred_candidate_takes_each_publications_current_answer() { + use PaymentApplication::{AwaitingSpendHistory, AwaitingWitness}; + for (deferral, answer) in [ + (AwaitingSpendHistory, Answer::Corrected), + (AwaitingSpendHistory, Answer::Unchanged), + (AwaitingSpendHistory, Answer::Dropped), + (AwaitingWitness, Answer::Dropped), + ] { + let mut r = restored(&[0]); + let (record, through) = (r.records[0].clone(), r.through); + let mut first = record.clone(); + if answer == Answer::Corrected { + // Publication A misstates the payment's transaction. + first.payment.tx_index += 1; + } + let wipe = match deferral { + AwaitingWitness => "DELETE FROM ironwood_tree_checkpoints", + _ => "DELETE FROM nullifier_map; DELETE FROM ironwood_nullifier_scan_blocks;", + }; + r.st.wallet().conn().execute_batch(wipe).unwrap(); + let mut notes = r.notes(); + let a = serve(&[first], through); + let swept = run(&mut r.st, through, &a, &mut notes, &NoLock, NOW) + .await + .unwrap(); + assert_matches!(swept.deferred[..], [(k, Error::Queued(d))] if k.index() == 0 && *d == deferral); + assert!(swept.pending); + + // Publication B, after the attempt's backoff. + let b = match answer { + Answer::Dropped => { + Directory::new(publish(&[], &[record.payment.cmx], through, &provider(&[]))) + } + _ => serve(&[record], through), + }; + if deferral == AwaitingSpendHistory { + run_clean(&mut r.st, through, &b, &mut notes, NOW + 60).await; + assert_eq!(lookups(&b), 0, "{:?}", b.requests()); + scan_queued(&mut r.st); + } + run_clean(&mut r.st, through, &b, &mut notes, NOW + 60).await; + settle(&mut r.st, through, &b, &mut notes).await; + let case = format!("{deferral:?} {answer:?}"); + let value = if answer == Answer::Dropped { 0 } else { VALUE }; + assert_eq!(balance(&r.st), value, "{case}"); + assert_eq!(lookups(&b) > 0, answer != Answer::Dropped, "{case}"); + let requested = if answer == Answer::Corrected { 2 } else { 1 }; + assert_eq!(notes.requested.len(), requested, "{case}"); + } +}