path: root/eth-wire/src/loops/worker.rs

/*
  This file is part of TALER
  Copyright (C) 2022 Taler Systems SA

  TALER is free software; you can redistribute it and/or modify it under the
  terms of the GNU Affero General Public License as published by the Free Software
  Foundation; either version 3, or (at your option) any later version.

  TALER is distributed in the hope that it will be useful, but WITHOUT ANY
  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
  A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.

  You should have received a copy of the GNU Affero General Public License along with
  TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
*/
use std::{fmt::Write, time::SystemTime};

use common::{
    api_common::base32,
    log::log::{error, info, warn},
    metadata::{InMetadata, OutMetadata},
    postgres::{fallible_iterator::FallibleIterator, Client},
    reconnect::AutoReconnectDb,
    sql::{sql_array, sql_url},
    status::{BounceStatus, DebitStatus},
};
use eth_wire::{
    rpc::{self, AutoRpcWallet, Rpc, RpcClient, Transaction, TransactionRequest},
    taler_util::{eth_payto_url, eth_to_taler},
    ListSinceSync, RpcExtended, SyncState, SyncTransaction,
};
use ethereum_types::{Address, H256, U256};

use crate::{
    fail_point::fail_point,
    loops::LoopError,
    sql::{sql_addr, sql_eth_amount, sql_hash},
    WireState,
};

use super::{analysis::analysis, LoopResult};

pub fn worker(mut rpc: AutoRpcWallet, mut db: AutoReconnectDb, mut state: WireState) {
    let mut lifetime = state.lifetime;
    let mut status = true;
    let mut skip_notification = false;

    loop {
        // Check lifetime
        if let Some(nb) = lifetime.as_mut() {
            if *nb == 0 {
                info!("Reach end of lifetime");
                return;
            } else {
                *nb -= 1;
            }
        }

        // Connect
        let rpc = rpc.client();
        let db = db.client();

        let result: LoopResult<()> = (|| {
            // Listen to all channels
            db.batch_execute("LISTEN new_block; LISTEN new_tx")?;
            // Wait for the next notification
            {
                let mut ntf = db.notifications();
                if !skip_notification && ntf.is_empty() {
                    // Block until next notification
                    ntf.blocking_iter().next()?;
                }
                // Conflate all notifications
                let mut iter = ntf.iter();
                while iter.next()?.is_some() {}
            }

            // It is not possible to atomically update the blockchain and the database.
            // When we failed to sync the database and the blockchain state we rely on
            // sync_chain to recover the lost updates.
            // When this function is running concurrently, it not possible to known another
            // execution has failed, and this can lead to a transaction being sent multiple time.
            // To ensure only a single version of this function is running at a given time we rely
            // on postgres advisory lock

            // Take the lock
            let row = db.query_one("SELECT pg_try_advisory_lock(42)", &[])?;
            let locked: bool = row.get(0);
            if !locked {
                return Err(LoopError::Concurrency);
            }

            // Get stored sync state
            let row = db.query_one("SELECT value FROM state WHERE name='sync'", &[])?;
            let sync_state = SyncState::from_bytes(&sql_array(&row, 0));

            // Get changes
            let list = rpc.list_since_sync(&state.address, sync_state, state.confirmation)?;

            // Perform analysis
            state.confirmation =
                analysis(list.fork_len, state.confirmation, state.max_confirmations)?;

            // Sync chain
            if sync_chain(db, &state, &mut status, list)? {
                // As we are now in sync with the blockchain if a transaction has Requested status it have not been sent

                // Send requested debits
                while debit(db, rpc, &state)? {}

                // Bump stuck transactions
                while bump(db, rpc, &state)? {}

                // Send requested bounce
                while bounce(db, rpc, state.bounce_fee)? {}
            }
            Ok(())
        })();

        if let Err(e) = result {
            error!("worker: {}", e);
            // When we catch an error, we sometimes want to retry immediately (eg. reconnect to RPC or DB).
            // Rpc error codes are generic. We need to match the msg to get precise ones. Some errors
            // can resolve themselves when a new block is mined (new fees, new transactions). Our simple
            // approach is to wait for the next loop when an RPC error is caught to prevent endless logged errors.
            skip_notification = !matches!(
                e,
                LoopError::Rpc(rpc::Error::RPC { .. }) | LoopError::Concurrency
            );
        } else {
            skip_notification = false;
        }
    }
}

/// Parse new transactions, return true if the database is up to date with the latest mined block
fn sync_chain(
    db: &mut Client,
    state: &WireState,
    status: &mut bool,
    list: ListSinceSync,
) -> LoopResult<bool> {
    // Get the current confirmation delay
    let conf_delay = state.confirmation;

    // Check if a confirmed incoming transaction have been removed by a blockchain reorganization
    let new_status = sync_chain_removed(&list.txs, &list.removed, db, &state.address, conf_delay)?;

    // Sync status with database
    if *status != new_status {
        let mut tx = db.transaction()?;
        tx.execute(
            "UPDATE state SET value=$1 WHERE name='status'",
            &[&[new_status as u8].as_ref()],
        )?;
        tx.execute("NOTIFY status", &[])?;
        tx.commit()?;
        *status = new_status;
        if new_status {
            info!("Recovered lost transactions");
        }
    }
    if !new_status {
        return Ok(false);
    }

    for sync_tx in list.txs {
        let tx = &sync_tx.tx;
        if tx.to == Some(state.address) && sync_tx.confirmations >= conf_delay {
            sync_chain_incoming_confirmed(tx, db, state)?;
        } else if tx.from == Some(state.address) {
            sync_chain_outgoing(&sync_tx, db, state)?;
        }
    }

    db.execute(
        "UPDATE state SET value=$1 WHERE name='sync'",
        &[&list.state.to_bytes().as_ref()],
    )?;
    Ok(true)
}

/// Sync database with removed transactions, return false if bitcoin backing is compromised
fn sync_chain_removed(
    txs: &[SyncTransaction],
    removed: &[SyncTransaction],
    db: &mut Client,
    addr: &Address,
    min_confirmation: u32,
) -> LoopResult<bool> {
    // A removed incoming transaction is a correctness issues in only two cases:
    // - it is a confirmed credit registered in the database
    // - it is an invalid transactions already bounced
    // Those two cases can compromise bitcoin backing
    // Removed outgoing transactions will be retried automatically by the node

    let mut blocking_credit = Vec::new();
    let mut blocking_bounce = Vec::new();

    // Only keep incoming transaction that are not reconfirmed
    // TODO study risk of accepting only mined transactions for faster recovery
    for tx in removed
        .iter()
        .filter(|sync_tx| {
            sync_tx.tx.to == Some(*addr)
                && txs
                    .iter()
                    .all(|it| it.tx.hash != sync_tx.tx.hash || it.confirmations < min_confirmation)
        })
        .map(|s| &s.tx)
    {
        match InMetadata::decode(&tx.input) {
            Ok(metadata) => match metadata {
                InMetadata::Credit { reserve_pub } => {
                    // Credits are only problematic if not reconfirmed and stored in the database
                    if db
                        .query_opt(
                            "SELECT 1 FROM tx_in WHERE reserve_pub=$1",
                            &[&reserve_pub.as_ref()],
                        )?
                        .is_some()
                    {
                        blocking_credit.push((reserve_pub, tx.hash, tx.from.unwrap()));
                    }
                }
            },
            Err(_) => {
                // Invalid tx are only problematic if if not reconfirmed and already bounced
                if let Some(row) = db.query_opt(
                    "SELECT txid FROM bounce WHERE bounced=$1 AND txid IS NOT NULL",
                    &[&tx.hash.as_ref()],
                )? {
                    blocking_bounce.push((sql_hash(&row, 0), tx.hash));
                } else {
                    // Remove transaction from bounce table
                    db.execute("DELETE FROM bounce WHERE bounced=$1", &[&tx.hash.as_ref()])?;
                }
            }
        }
    }

    if !blocking_bounce.is_empty() || !blocking_credit.is_empty() {
        let mut buf = "The following transaction have been removed from the blockchain, ethereum backing is compromised until the transaction reappear:".to_string();
        for (key, id, addr) in blocking_credit {
            write!(
                &mut buf,
                "\n\tcredit {} in {} from {}",
                base32(&key),
                hex::encode(id),
                hex::encode(addr)
            )
            .unwrap();
        }
        for (id, bounced) in blocking_bounce {
            write!(
                &mut buf,
                "\n\tbounce {} in {}",
                hex::encode(id),
                hex::encode(bounced)
            )
            .unwrap();
        }
        error!("{}", buf);
        Ok(false)
    } else {
        Ok(true)
    }
}

/// Sync database with an incoming confirmed transaction
fn sync_chain_incoming_confirmed(
    tx: &Transaction,
    db: &mut Client,
    state: &WireState,
) -> Result<(), LoopError> {
    match InMetadata::decode(&tx.input) {
        Ok(metadata) => match metadata {
            InMetadata::Credit { reserve_pub } => {
                let date = SystemTime::now();
                let amount = eth_to_taler(&tx.value, state.currency);
                let credit_addr = tx.from.expect("Not coinbase");
                let nb = db.execute("INSERT INTO tx_in (_date, amount, reserve_pub, debit_acc, credit_acc) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (reserve_pub) DO NOTHING ", &[
                &date, &amount.to_string(), &reserve_pub.as_ref(), &eth_payto_url(&credit_addr).as_ref(), &state.payto.as_ref()
            ])?;
                if nb > 0 {
                    info!(
                        "<< {} {} in {} from {}",
                        amount,
                        base32(&reserve_pub),
                        hex::encode(tx.hash),
                        hex::encode(credit_addr),
                    );
                }
            }
        },
        Err(_) => {
            // If encoding is wrong request a bounce
            db.execute(
                "INSERT INTO bounce (bounced) VALUES ($1) ON CONFLICT (bounced) DO NOTHING",
                &[&tx.hash.as_ref()],
            )?;
        }
    }
    Ok(())
}

/// Sync database with an outgoing transaction
fn sync_chain_outgoing(tx: &SyncTransaction, db: &mut Client, state: &WireState) -> LoopResult<()> {
    let SyncTransaction { tx, confirmations } = tx;
    match OutMetadata::decode(&tx.input) {
        Ok(metadata) => match metadata {
            OutMetadata::Debit { wtid, .. } => {
                let amount = eth_to_taler(&tx.value, state.currency);
                let credit_addr = tx.to.unwrap();
                // Get previous out tx
                let row = db.query_opt(
                    "SELECT id, status, sent FROM tx_out WHERE wtid=$1 FOR UPDATE",
                    &[&wtid.as_ref()],
                )?;
                if let Some(row) = row {
                    // If already in database, sync status
                    let row_id: i32 = row.get(0);
                    let status: i16 = row.get(1);
                    let sent: Option<SystemTime> = row.get(2);

                    let expected_status = DebitStatus::Sent as i16;
                    let expected_send = sent.filter(|_| *confirmations == 0);
                    if status != expected_status || sent != expected_send {
                        let nb_row = db.execute(
                        "UPDATE tx_out SET status=$1, txid=$2, sent=NULL WHERE id=$3 AND status=$4",
                        &[
                            &(DebitStatus::Sent as i16),
                            &tx.hash.as_ref(),
                            &row_id,
                            &status,
                        ],
                    )?;
                        if nb_row > 0 {
                            match DebitStatus::try_from(status as u8).unwrap() {
                                DebitStatus::Requested => {
                                    warn!(
                                        ">> (recovered) {} {} in {} to {}",
                                        amount,
                                        base32(&wtid),
                                        hex::encode(tx.hash),
                                        hex::encode(credit_addr)
                                    );
                                }
                                DebitStatus::Sent => { /* Status is correct */ }
                            }
                        }
                    }
                } else {
                    // Else add to database
                    let date = SystemTime::now();
                    let nb = db.execute(
                    "INSERT INTO tx_out (_date, amount, wtid, debit_acc, credit_acc, exchange_url, status, txid, request_uid) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (wtid) DO NOTHING",
                    &[&date, &amount.to_string(), &wtid.as_ref(), &eth_payto_url(&state.address).as_ref(), &eth_payto_url(&credit_addr).as_ref(), &state.base_url.as_ref(), &(DebitStatus::Sent as i16), &tx.hash.as_ref(), &None::<&[u8]>],
                        )?;
                    if nb > 0 {
                        warn!(
                            ">> (onchain) {} {} in {} to {}",
                            amount,
                            base32(&wtid),
                            hex::encode(tx.hash),
                            hex::encode(credit_addr)
                        );
                    }
                }
            }
            OutMetadata::Bounce { bounced } => {
                let bounced = H256::from_slice(&bounced);
                // Get previous bounce
                let row = db.query_opt(
                    "SELECT id, status FROM bounce WHERE bounced=$1",
                    &[&bounced.as_ref()],
                )?;
                if let Some(row) = row {
                    // If already in database, sync status
                    let row_id: i32 = row.get(0);
                    let status: i16 = row.get(1);
                    match BounceStatus::try_from(status as u8).unwrap() {
                        BounceStatus::Requested => {
                            let nb_row = db.execute(
                                "UPDATE bounce SET status=$1, txid=$2 WHERE id=$3 AND status=$4",
                                &[
                                    &(BounceStatus::Sent as i16),
                                    &tx.hash.as_ref(),
                                    &row_id,
                                    &status,
                                ],
                            )?;
                            if nb_row > 0 {
                                warn!(
                                    "|| (recovered) {} in {}",
                                    hex::encode(bounced),
                                    hex::encode(tx.hash)
                                );
                            }
                        }
                        BounceStatus::Ignored => error!(
                            "watcher: ignored bounce {} found in chain at {}",
                            bounced,
                            hex::encode(tx.hash)
                        ),
                        BounceStatus::Sent => { /* Status is correct */ }
                    }
                } else {
                    // Else add to database
                    let nb = db.execute(
        "INSERT INTO bounce (bounced, txid, status) VALUES ($1, $2, $3) ON CONFLICT (txid) DO NOTHING",
        &[&bounced.as_ref(), &tx.hash.as_ref(), &(BounceStatus::Sent as i16)],
            )?;
                    if nb > 0 {
                        warn!(
                            "|| (onchain) {} in {}",
                            hex::encode(bounced),
                            hex::encode(tx.hash)
                        );
                    }
                }
            }
        },
        Err(_) => { /* Ignore */ }
    }
    Ok(())
}

/// Send a debit transaction on the blockchain, return false if no more requested transactions are found
fn debit(db: &mut Client, rpc: &mut Rpc, state: &WireState) -> LoopResult<bool> {
    // We rely on the advisory lock to ensure we are the only one sending transactions
    let row = db.query_opt(
"SELECT id, amount, wtid, credit_acc, exchange_url FROM tx_out WHERE status=$1 ORDER BY _date LIMIT 1",
&[&(DebitStatus::Requested as i16)],
)?;
    if let Some(row) = &row {
        let id: i32 = row.get(0);
        let amount = sql_eth_amount(row, 1, state.currency);
        let wtid: [u8; 32] = sql_array(row, 2);
        let addr = sql_addr(row, 3);
        let url = sql_url(row, 4);
        let tx_id = rpc.debit(state.address, addr, amount, wtid, url)?;
        fail_point("(injected) fail debit", 0.3)?;
        db.execute(
            "UPDATE tx_out SET status=$1, txid=$2, sent=now() WHERE id=$3",
            &[&(DebitStatus::Sent as i16), &tx_id.as_ref(), &id],
        )?;
        let amount = eth_to_taler(&amount, state.currency);
        info!(
            ">> {} {} in {} to {}",
            amount,
            base32(&wtid),
            hex::encode(tx_id),
            hex::encode(addr)
        );
    }
    Ok(row.is_some())
}

/// Bump a stuck transaction, return false if no more stuck transactions are found
fn bump(db: &mut Client, rpc: &mut Rpc, state: &WireState) -> LoopResult<bool> {
    if let Some(delay) = state.bump_delay {
        // We rely on the advisory lock to ensure we are the only one sending transactions
        let row = db.query_opt(
        "SELECT id, txid FROM tx_out WHERE status=$1 AND EXTRACT(EPOCH FROM (now() - sent)) > $2 ORDER BY _date LIMIT 1",
        &[&(DebitStatus::Sent as i16), &(delay as f64)],
        )?;
        if let Some(row) = &row {
            let id: i32 = row.get(0);
            let txid = sql_hash(row, 1);
            let tx = rpc.get_transaction(&txid)?.expect("Bump existing tx");
            rpc.send_transaction(&TransactionRequest {
                from: tx.from.unwrap(),
                to: tx.to.unwrap(),
                value: tx.value,
                gas_price: None,
                data: tx.input,
                nonce: Some(tx.nonce),
            })?;
            let row = db.query_one(
                "UPDATE tx_out SET sent=now() WHERE id=$1 RETURNING wtid",
                &[&id],
            )?;
            info!(">> (bump) {} in {}", base32(row.get(0)), hex::encode(txid));
        }
        Ok(row.is_some())
    } else {
        Ok(false)
    }
}

/// Bounce a transaction on the blockchain, return false if no more requested transactions are found
fn bounce(db: &mut Client, rpc: &mut Rpc, fee: U256) -> LoopResult<bool> {
    // We rely on the advisory lock to ensure we are the only one sending transactions
    let row = db.query_opt(
        "SELECT id, bounced FROM bounce WHERE status=$1 ORDER BY _date LIMIT 1",
        &[&(BounceStatus::Requested as i16)],
    )?;
    if let Some(row) = &row {
        let id: i32 = row.get(0);
        let bounced: H256 = sql_hash(row, 1);

        let bounce = rpc.bounce(bounced, fee)?;
        match bounce {
            Some(hash) => {
                fail_point("(injected) fail bounce", 0.3)?;
                db.execute(
                    "UPDATE bounce SET txid=$1, status=$2 WHERE id=$3",
                    &[&hash.as_ref(), &(BounceStatus::Sent as i16), &id],
                )?;
                info!("|| {} in {}", hex::encode(bounced), hex::encode(hash));
            }
            None => {
                db.execute(
                    "UPDATE bounce SET status=$1 WHERE id=$2",
                    &[&(BounceStatus::Ignored as i16), &id],
                )?;
                info!("|| (ignore) {} ", hex::encode(bounced));
            }
        }
    }
    Ok(row.is_some())
}