feat(asb): async swap setup

[binarybaron]

Note

Medium Risk
Refactors swap setup handling to run wallet snapshot computation asynchronously via a new FuturesUnordered pipeline, which could affect timing/error handling and introduce subtle concurrency or dropped-response edge cases.

Overview
Swap setup is now non-blocking. The event loop no longer computes the wallet snapshot inline when SwapSetupInitiated fires; it instead queues an async task in a new inflight_wallet_snapshots stream and keeps the loop responsive.

When a queued snapshot task completes, the event loop then applies the anti-spam/refund policy and responds back to the swap setup handler with (WalletSnapshot, amnesty_amount, should_publish_tx_withhold), adding logging for failures and dropped connections.

^{Written by Cursor Bugbot for commit 5251435. This will update automatically on new commits. Configure here.}

[binarybaron]

bugbot run

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

[Einliterflasche]

Why did you split the logic into two steps (wait for requests -> process request)? We can also put them into the same future and avoid the inflight_wallet_snapshots_requests buffer.

self.inflight_wallet_snapshots.push(async move {
    let (btc, responder) = send_wallet_snapshot.recv().await?;
    return Ok((responder, capture_wallet_snapshot().await?))
})

or something like that.

[binarybaron]

Why did you split the logic into two steps (wait for requests -> process request)? We can also put them into the same future and avoid the inflight_wallet_snapshots_requests buffer.

self.inflight_wallet_snapshots.push(async move {
    let (btc, responder) = send_wallet_snapshot.recv().await?;
    return Ok((responder, capture_wallet_snapshot().await?))
})

or something like that.

Done

[binarybaron]

bugbot run

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

[binarybaron]

@Einliterflasche Please review again.

[Einliterflasche]

Next logic though is why we would bother doing the final calculations in the event loop as well if we can just do everything in the future.

self.inflight_wallet_snapshots.push(async move {
    let (btc, responder) = send_wallet_snapshot.recv().await?;
    let snapshot = capture_wallet_snapshot(btc).await?;
    let (anti_spam_stuff, ...) = apply_refund_policy(btc);
    responder.respond(snapshot, anti_spam_stuff).await?;
    Ok(())
})

Might be too implicit. idk. Apart from that it LGTM (altough I haven't run the tests)

[binarybaron]

Next logic though is why we would bother doing the final calculations in the event loop as well if we can just do everything in the future.

self.inflight_wallet_snapshots.push(async move {
    let (btc, responder) = send_wallet_snapshot.recv().await?;
    let snapshot = capture_wallet_snapshot(btc).await?;
    let (anti_spam_stuff, ...) = apply_refund_policy(btc);
    responder.respond(snapshot, anti_spam_stuff).await?;
    Ok(())
})

Might be too implicit. idk. Apart from that it LGTM (altough I haven't run the tests)

I think this is too implicit.

[Einliterflasche]

Next logic though is why we would bother doing the final calculations in the event loop as well if we can just do everything in the future.

self.inflight_wallet_snapshots.push(async move {
    let (btc, responder) = send_wallet_snapshot.recv().await?;
    let snapshot = capture_wallet_snapshot(btc).await?;
    let (anti_spam_stuff, ...) = apply_refund_policy(btc);
    responder.respond(snapshot, anti_spam_stuff).await?;
    Ok(())
})

Might be too implicit. idk. Apart from that it LGTM (altough I haven't run the tests)

I think this is too implicit.

Then I think this is good to merge (assuming tests pass).