acta_maker_sdk/ws/vault/
sell.rs

1use std::future::Future;
2use std::time::Duration;
3
4use base64::Engine as _;
5use base64::engine::general_purpose::STANDARD;
6use solana_sdk::pubkey::Pubkey;
7use uuid::Uuid;
8
9use crate::chain::vault::{ComputeBudgetBounds, Counterparty, VaultManager, VaultTradeTerms};
10use crate::chain::{ChainClient, ChainError, MarketInfo};
11use crate::types::ids::OrderId;
12use crate::types::{PositionType, Quantity, Strike, TimeoutSeconds};
13use crate::ws::managed::ManagedReceiveError;
14use crate::ws::types::{
15    AcceptQuoteMessage, ClientMessage, Delivery, QuoteReceivedMessage, RfqRequestMessage,
16    ServerMessage, SponsoredTxToSignData,
17};
18
19use super::client::{VaultOperatorClient, VaultRequestError};
20use super::gate::SendGate;
21use super::tx::VaultTxError;
22
23/// An option the vault writes and sells through an RFQ.
24#[derive(Debug, Clone)]
25pub struct VaultSale {
26    pub market_pda: Pubkey,
27    pub market: MarketInfo,
28    pub strike: Strike,
29    pub quantity: Quantity,
30    pub rfq_timeout: TimeoutSeconds,
31    /// Bounds quote collection with `take`, and then serving plus `writer`, verification
32    /// and signing; RFQ, accept and submit commands are bounded by the request timeout.
33    pub window: Duration,
34    pub bounds: ComputeBudgetBounds,
35}
36
37/// The bot's decisions and bookkeeping inside [`VaultOperatorClient::sell_rfq`].
38pub trait SaleDesk: Send {
39    type Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static;
40
41    /// Whether to take `quote`; asked for each quote until one is taken.
42    fn take(
43        &mut self,
44        quote: &QuoteReceivedMessage,
45    ) -> impl Future<Output = Result<bool, Self::Error>> + Send;
46
47    /// Runs before `AcceptQuote` is sent for the taken quote.
48    fn accepting(&mut self, _quote: &QuoteReceivedMessage) -> Result<(), Self::Error> {
49        Ok(())
50    }
51
52    /// Re-checks the vault once the transaction is served and returns the writer to sign as.
53    fn writer(&mut self) -> impl Future<Output = Result<VaultManager, Self::Error>> + Send;
54
55    /// Runs with the signed transaction before it is submitted; an error leaves it unsubmitted.
56    fn signed(&mut self, _sale: &SignedSale) -> Result<(), Self::Error> {
57        Ok(())
58    }
59}
60
61/// A signed sale transaction; `signature` is its transaction id.
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct SignedSale {
64    pub order_id: OrderId,
65    pub signature: String,
66    pub last_valid_block_height: u64,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub enum SaleOutcome {
71    /// No quote was taken within the window.
72    Unfilled,
73    /// The server accepted the signed transaction; its chain outcome is pending.
74    Submitted(SignedSale),
75}
76
77#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78pub enum SaleStage {
79    Request,
80    Take,
81    Accept,
82    Sign,
83}
84
85#[derive(Debug, thiserror::Error)]
86#[non_exhaustive]
87pub enum SaleError<E> {
88    #[error("market_pda is not the program's market for these terms")]
89    Market,
90    #[error("sale desk: {0}")]
91    Desk(E),
92    #[error("{stage:?} failed")]
93    Rejected {
94        stage: SaleStage,
95        #[source]
96        error: VaultRequestError,
97    },
98    #[error(transparent)]
99    Stream(#[from] ManagedReceiveError),
100    #[error("the session changed during the sale")]
101    SessionChanged,
102    #[error("the accepted quote was not served within the window")]
103    Unserved,
104    /// `take`, or `writer` plus verification and signing, outlived the window.
105    #[error("{0:?} did not finish within the window")]
106    TimedOut(SaleStage),
107    #[error(transparent)]
108    Chain(#[from] ChainError),
109    #[error(transparent)]
110    Sign(#[from] VaultTxError),
111    #[error("submitting the signed sale failed")]
112    Submit {
113        sale: SignedSale,
114        #[source]
115        error: VaultRequestError,
116    },
117}
118
119impl<E> SaleError<E> {
120    /// The signed sale this failure may have left in flight. `None`: nothing
121    /// signed reached the server, so the sale cannot execute.
122    #[must_use]
123    pub fn in_flight(&self) -> Option<&SignedSale> {
124        match self {
125            Self::Submit { sale, error } if error.delivery() != Some(Delivery::NotForwarded) => {
126                Some(sale)
127            }
128            _ => None,
129        }
130    }
131}
132
133impl VaultOperatorClient {
134    /// Sells `sale` in `connection_epoch`: requests the RFQ, offers each quote to
135    /// `desk`, accepts the first it takes, verifies and signs the served
136    /// transaction against the chain counterparty, and submits it. Every send
137    /// goes through `gate`.
138    pub async fn sell_rfq<D: SaleDesk>(
139        &self,
140        chain: &ChainClient,
141        sale: &VaultSale,
142        connection_epoch: u64,
143        gate: &(impl SendGate + Sync),
144        desk: &mut D,
145    ) -> Result<SaleOutcome, SaleError<D::Error>> {
146        let market = &sale.market;
147        let canonical = crate::quote::market_pda(
148            &chain.program_id(),
149            &market.underlying_mint,
150            &market.quote_mint,
151            market.expiry_ts,
152            market.is_put,
153        );
154        if canonical != sale.market_pda {
155            return Err(SaleError::Market);
156        }
157        let mut messages = self.subscribe_messages();
158        let request_id = Uuid::new_v4();
159        self.gated_command(
160            gate,
161            connection_epoch,
162            ClientMessage::RfqRequest(RfqRequestMessage {
163                request_id,
164                market: sale.market_pda.to_string().into(),
165                position_type: if market.is_put {
166                    PositionType::CashSecuredPut
167                } else {
168                    PositionType::CoveredCall
169                },
170                strike: sale.strike,
171                quantity: sale.quantity,
172                timeout_seconds: sale.rfq_timeout,
173            }),
174        )
175        .await
176        .map_err(|error| SaleError::Rejected {
177            stage: SaleStage::Request,
178            error,
179        })?;
180
181        let mut rfq_id = None;
182        let mut taken: Option<QuoteReceivedMessage> = None;
183        let mut deadline = tokio::time::Instant::now() + sale.window;
184        loop {
185            let Ok(inbound) = tokio::time::timeout_at(deadline, messages.recv()).await else {
186                return match taken {
187                    None => Ok(SaleOutcome::Unfilled),
188                    Some(_) => Err(SaleError::Unserved),
189                };
190            };
191            let inbound = inbound?;
192            if inbound.connection_epoch != connection_epoch {
193                return Err(SaleError::SessionChanged);
194            }
195            match inbound.message() {
196                ServerMessage::RfqCreated(created) if created.request_id == Some(request_id) => {
197                    rfq_id = Some(created.rfq_id);
198                }
199                ServerMessage::QuoteReceived(quote) if taken.is_none() => {
200                    if Some(quote.rfq_id) == rfq_id
201                        && quote.strike == sale.strike
202                        && tokio::time::timeout_at(deadline, desk.take(quote))
203                            .await
204                            .map_err(|_| SaleError::TimedOut(SaleStage::Take))?
205                            .map_err(SaleError::Desk)?
206                    {
207                        desk.accepting(quote).map_err(SaleError::Desk)?;
208                        self.gated_command(
209                            gate,
210                            connection_epoch,
211                            ClientMessage::AcceptQuote(AcceptQuoteMessage {
212                                request_id: Uuid::new_v4(),
213                                rfq_id: quote.rfq_id,
214                                maker: quote.maker.clone(),
215                                order_id: quote.order_id,
216                            }),
217                        )
218                        .await
219                        .map_err(|error| SaleError::Rejected {
220                            stage: SaleStage::Accept,
221                            error,
222                        })?;
223                        taken = Some(quote.clone());
224                        deadline = tokio::time::Instant::now() + sale.window;
225                    }
226                }
227                ServerMessage::SponsoredTxToSign(served) => {
228                    if let Some(quote) = taken.as_ref().filter(|q| q.order_id == served.order_id) {
229                        let signed = tokio::time::timeout_at(
230                            deadline,
231                            self.sign_sale(chain, sale, quote, served, desk),
232                        )
233                        .await
234                        .map_err(|_| SaleError::TimedOut(SaleStage::Sign))??;
235                        return self.submit_sale(gate, connection_epoch, signed, desk).await;
236                    }
237                }
238                _ => {}
239            }
240        }
241    }
242
243    async fn sign_sale<D: SaleDesk>(
244        &self,
245        chain: &ChainClient,
246        sale: &VaultSale,
247        quote: &QuoteReceivedMessage,
248        served: &SponsoredTxToSignData,
249        desk: &mut D,
250    ) -> Result<(SignedSale, crate::ws::types::SubmitSignedSponsoredTxData), SaleError<D::Error>>
251    {
252        let writer = desk.writer().await.map_err(SaleError::Desk)?;
253        let maker = quote
254            .maker
255            .parse::<Pubkey>()
256            .map_err(|_| ChainError::Identity("quote maker"))?;
257        let counterparty = Counterparty::load(chain, maker).await?;
258        let terms = VaultTradeTerms {
259            underlying_mint: sale.market.underlying_mint,
260            quote_mint: sale.market.quote_mint,
261            expiry_ts: sale.market.expiry_ts,
262            is_put: sale.market.is_put,
263            strike: sale.strike.value(),
264            quantity: sale.quantity.value(),
265            price: quote.price.value(),
266            valid_until: quote.valid_until.as_unix_seconds(),
267            nonce: quote.nonce.value(),
268        };
269        let tables = chain
270            .address_lookup_tables(
271                &crate::chain::vault::served_lookup_table_keys(&served.tx_base64)
272                    .map_err(VaultTxError::from)?,
273            )
274            .await?;
275        let submit = self.verify_and_sign_sponsored_tx(
276            served,
277            &counterparty.sale(&writer, terms),
278            &tables,
279            sale.bounds,
280        )?;
281        let signed = SignedSale {
282            order_id: served.order_id,
283            signature: transaction_id(&submit.tx_base64)?,
284            last_valid_block_height: served.last_valid_block_height,
285        };
286        Ok((signed, submit))
287    }
288
289    async fn submit_sale<D: SaleDesk>(
290        &self,
291        gate: &(impl SendGate + Sync),
292        connection_epoch: u64,
293        (signed, submit): (SignedSale, crate::ws::types::SubmitSignedSponsoredTxData),
294        desk: &mut D,
295    ) -> Result<SaleOutcome, SaleError<D::Error>> {
296        desk.signed(&signed).map_err(SaleError::Desk)?;
297        match self
298            .gated_command(
299                gate,
300                connection_epoch,
301                ClientMessage::SubmitSignedSponsoredTx(submit),
302            )
303            .await
304        {
305            Ok(()) => Ok(SaleOutcome::Submitted(signed)),
306            Err(error) => Err(SaleError::Submit {
307                sale: signed,
308                error,
309            }),
310        }
311    }
312}
313
314/// First signature of a signed wire transaction, base58.
315fn transaction_id(tx_base64: &str) -> Result<String, VaultTxError> {
316    let malformed = || {
317        VaultTxError::Intent(crate::chain::vault::VaultIntentError::Malformed(
318            "signed transaction",
319        ))
320    };
321    let bytes = STANDARD.decode(tx_base64).map_err(|_| malformed())?;
322    match bytes.first() {
323        Some(count) if (1..0x80).contains(count) => bytes
324            .get(1..65)
325            .map(crate::wire::encode_base58)
326            .ok_or_else(malformed),
327        _ => Err(malformed()),
328    }
329}
acta_maker_sdk/ws/vault/sell.rs

acta_maker_sdk/ws/vault/
sell.rs
